LangGraph 是什么？
LangGraph 是 LangChain 生态下的核心库，专门用于构建有状态、可循环、支持分支 / 并发的复杂工作流（比如智能体 Agent、多智能体协作、任务拆解与执行等）。
在看到上面任务拆解与执行等功能的时候，你可能会疑惑，在langchain中也提供了tools用于条件判断，但是tools是通过大模型本身来判断：
from langchain.tools import StructuredTool
# 步骤 1：定义工具函数（普通 Python 函数）
def calculate_expression(expression: str) -> str:
"""
用于计算数学表达式的工具，支持加减乘除、小数运算。
参数：
expression: 字符串格式的数学表达式，比如 "100-23.5"、"5*6+8"
返回：
计算结果（字符串格式）
"""
try:
# 注意：生产环境中避免用 eval，此处仅为示例，可替换为安全的表达式解析库（如 ast）
result = eval(expression)
return f"计算结果：{result}"
except Exception as e:
return f"计算失败：{str(e)}"
# 步骤 2：将函数包装为 LangChain Tool（核心：让 Agent 识别工具）
# StructuredTool 支持结构化参数（更规范），Tool 是简化版
calculator_tool = StructuredTool.from_function(
func=calculate_expression, # 绑定的函数
name="calculator", # 工具名（必须唯一）
description="用于计算数学表达式的工具，当用户问题涉及加减乘除、小数运算时调用此工具", # 关键：让 Agent 判断何时调用
)
这样的判断在实际生产项目中可能会出现误差，有时候可以有时候又不可以的情形，同时在调试的时候也比较麻烦，而LangGraph中支持你自己定义判断。
LangGraph 核心概念（从零理解）
先记住 4 个核心概念，所有 LangGraph 开发都围绕它们展开：
State（状态）: 图的 “数据容器”，所有节点共享 / 修改这个状态（比如用户问题、工具结果、回答）
Node（节点）: 执行具体操作的单元（比如调用 LLM、调用工具、格式化结果）
Edge（边）: 定义节点之间的连接关系（普通边：直接跳转；条件边：根据状态决定跳转方向）
Graph（图）: 整合所有节点、边、状态的容器，定义整个工作流的执行逻辑
一个简单的例子：
from typing import TypedDict, Optional
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
# 初始化大语言模型
llm = ChatOpenAI(model="gpt-3.5-turbo")
# 1. 扩展状态（增加工具调用相关字段）
class AgentState(TypedDict):
question: str
need_tool: Optional[bool] # 是否需要调用工具
tool_result: Optional[str] # 工具调用结果
final_answer: str # 最终回答
# 2. 节点1：判断是否需要调用工具
def check_tool_need(state: AgentState) -> AgentState:
question = state["question"]
# 简单判断：包含"天气"则需要调用工具
need_tool = "天气" in question
return {"need_tool": need_tool}
# 节点2：调用工具（模拟获取天气）
def call_weather_tool(state: AgentState) -> AgentState:
question = state["question"]
# 模拟工具调用结果（实际开发中替换为真实工具）
tool_result = f"【天气工具结果】北京今日气温：5℃，晴"
return {"tool_result": tool_result}
# 节点3：生成最终回答（分有无工具结果两种情况）
def generate_final_answer(state: AgentState) -> AgentState:
question = state["question"]
if state["need_tool"] and state["tool_result"]:
answer = f"你的问题：{question}\n工具结果：{state['tool_result']}\n最终回答：{state['tool_result']}"
else:
answer = llm.invoke(f"回答问题：{question}").content
return {"final_answer": answer}
# 3. 定义条件分支函数（返回下一个节点名称）
def decide_next_node(state: AgentState) -> str:
if state["need_tool"]:
return "call_tool" # 需要工具则跳转到调用工具节点
else:
return "generate_answer" # 不需要则直接生成回答
# 4. 构建图
def create_graph():
workflow = StateGraph(AgentState)
# 添加节点
workflow.add_node("check_tool", check_tool_need)
workflow.add_node("call_tool", call_weather_tool)
workflow.add_node("generate_answer", generate_final_answer)
# 设置入口
workflow.add_edge(START, "check_tool")
# 添加条件边
workflow.add_conditional_edges(
"check_tool", # 从check_tool节点出发
decide_next_node, # 根据此函数决定下一步
{
"call_tool": "call_tool", # 如果函数返回"call_tool"，则跳转到call_tool节点
"generate_answer": "generate_answer" # 如果函数返回"generate_answer"，则跳转到generate_answer节点
}
)
# 添加普通边
workflow.add_edge("call_tool", "generate_answer") # 调用工具后进入生成答案节点
workflow.add_edge("generate_answer", END) # 生成答案后结束
# 编译图
graph = workflow.compile()
return graph
这个例子中，可能对于初学者来说比较难以理解的是边的添加，特别是条件边。
边（Edge）：只负责 “指挥流程走向” 的规则
边是 LangGraph 工作流的 “指挥官”，它不做任何实际操作，只定义 “上一个节点做完后，下一个该执行哪个节点”。
LangGraph 里有两种边，对应两种流程走向：
1. 普通边（固定走向）：一条路走到黑
用 add_edge(源节点, 目标节点) 定义，意思是 “源节点做完，必须去目标节点”，没有任何条件
案例里的普通边：
# 调用工具节点做完，必须去生成回答节点
graph_builder.add_edge("call_tool", "generate_answer")
# 生成回答节点做完，必须结束流程（END是内置终点）
graph_builder.add_edge("generate_answer", END)
类比流水线：“组装工位做完，必须送到包装工位”，没有选择。
2. 条件边（动态走向）：根据情况选路走
用 add_conditional_edges 定义，意思是 “源节点做完后，根据 State 的内容，决定去哪个目标节点”。
这是 LangGraph 最核心的能力，也是边的灵魂。
案例里的条件边：
graph_builder.add_conditional_edges(
"check_tool", # 源节点：检查工具需求的节点
decide_next_node, # 判断函数：根据State选下一个节点
{"call_tool": "call_tool", "generate_answer": "generate_answer"} # 映射关系
)
- 判断函数
decide_next_node 会读取 State 里的 need_tool 字段； - 如果
need_tool=True，就去 call_tool 节点； - 如果
need_tool=False，就去 generate_answer 节点。
2026.01.13 下午
信息家