常见的大模型回答问题,往往是单论交互:输入问题—->模型输出结果,End。
而智能体应用,需要一个多步骤、有状态、会决策的流程,例如:客服智能体:一般要根据用户输入的订单号,先查询订单信息,并根据具体情况转人工或者走退款审批流程。这类场景的特点是:
1、有明确的步骤顺序:先执行A操作,再根据结果决定下一步。
2、存在分支和循环,查不到信息可以执行重试,权限不足转人工等等。
3、需要记忆与状态。
LangGraph:专门为构造这种“有状态、多步骤、可控的智能体”而产生的框架。是一个用“图”的方式来定义和运行智能体工作流的库。智能体被建模成一个有节点(Node)和边(Edge)的有向图,每个节点是一个计算步骤,边则控制数据流动的下一步走向。langGraph使用的图是数学分支——图论(Graph Theory)理论体系下的产物。
核心概念(用最简单的比喻)
想象你在做一个“请假审批”的流程图:
节点(Node):一个处理步骤,比如“主管审批”、“人事备案”。
在 LangGraph 中,节点可以是调用大模型的函数、调用工具的函数、或是纯粹的逻辑判断。
边(Edge):连接节点的箭头,表示“下一步去哪儿”。边可以是固定的,也可以是根据条件动态选择的。
状态(State):整个流程共享的一个数据对象(通常是一个字典),就像流程图中流转的“表单”。每个节点都可以读取和更新这个状态。
图(Graph):由节点和边组成的完整工作流结构,你可以把它编译成一个可运行的程序。
为什么选择 LangGraph?(相比自己写循环)
支持循环和条件分支:你自己写 while 循环很容易失控(无限循环、状态丢失),而 LangGraph 通过图编译,能自动管理执行流,还支持设置递归上限。
持久化和断点恢复:LangGraph 能一行代码就实现“检查点”,让智能体流程在任意节点中断后,重新运行时从断点继续。这在需要“等待人类批准”的场景里非常关键。
流式输出:天然支持逐步输出每个节点的结果,方便构建实时交互的应用。
可视化和调试:图结构本身就是文档,而且可以用 LangSmith 等工具调试流程。
一个极简示例:
定义class:
from typing import TypedDict
class AgentState(TypedDict):
messages: list # 存放历史对话列表
ask_again: bool # 控制是否再问一次
定义节点:
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
# 节点1:生成初次回复
def generate_first_reply(state: AgentState):
response = model.invoke(state["messages"])
# 简单的条件:如果回复里有“?”就标记再次询问
ask = "?" in response.content
return {"messages": [response], "ask_again": ask}
# 节点2:在条件满足时,基于前面内容再追问一个后续问题
def ask_followup(state: AgentState):
# 将助手的上一条回复和“请追问一个相关问题”加入提示
last_msg = state["messages"][-1]
prompt = f"Based on: {last_msg.content}\\nAsk an interesting follow-up question:"
response = model.invoke(prompt)
return {"messages": [response], "ask_again": False}
构建图:
from langgraph.graph import StateGraph, END
graph = StateGraph(AgentState)
# 添加节点
graph.add_node("first_reply", generate_first_reply)
graph.add_node("followup", ask_followup)
# 设置入口
graph.set_entry_point("first_reply")
# 添加条件边:根据 ask_again 决定去 followup 还是结束
def should_continue(state: AgentState):
if state["ask_again"]:
return "followup"
return END
graph.add_conditional_edges("first_reply", should_continue)
graph.add_edge("followup", END) # followup 节点后直接结束
# 编译成可运行应用
app = graph.compile()
调用方:
inputs = {"messages": ["Hello, how are you?"]}
for output in app.stream(inputs):
# 每个节点的输出都会流式打印
print(output)




