引言:从链式思维到图式思维
在构建AI驱动的应用程序时,我们早已超越了简单的“输入-输出”模型。现实世界的任务往往是多步骤、有状态且充满分支决策的。传统的LangChain通过“链”(Chain)将多个组件串联起来,虽然解决了顺序执行的问题,但在处理循环、条件分支和长期状态管理时,仍显得力不从心。
LangGraph应运而生。它不是一个替代品,而是LangChain生态中的一个强大扩展,将“链”的思想提升到了“图”(Graph)的维度。它允许开发者将AI应用的工作流建模为一张有向图,其中节点代表执行单元(可以是LLM调用、工具调用或自定义函数),边则定义了节点间的流转逻辑。这为构建智能体(Agent)、复杂对话系统、工作流引擎等应用提供了前所未有的灵活性和控制力。
本文将带你深入理解LangGraph的核心概念、基本用法,并通过实战示例展示如何用它构建一个具备记忆和工具调用能力的智能体。
核心概念
在深入代码之前,让我们先厘清LangGraph中的几个关键概念:
State(状态):这是贯穿整个图执行过程的共享上下文,通常是一个字典(
TypedDict)。所有节点都可以读取和修改状态中的字段。状态是LangGraph实现有状态、多轮交互的核心。Node(节点):图中的一个执行单元。每个节点都是一个函数,它接收当前的
State作为输入,并返回一个对该State的更新。节点可以执行任何操作:调用LLM、运行工具、进行条件判断、操作数据库等。Edge(边):定义节点之间如何连接和流转。边分为两种:
- 条件边(Conditional Edge):根据
State中的某个值(通常是上一个节点输出的结果)来决定下一步走向哪个节点。这实现了if-else分支逻辑。 - 普通边:无条件地从一个节点指向下一个节点。
- 条件边(Conditional Edge):根据
Graph(图):由节点和边组成的整体结构。LangGraph负责管理状态的传递、节点的调度以及循环的控制。
快速开始:构建你的第一个LangGraph应用
让我们从一个最简单的例子开始:一个接收用户问题,并调用LLM进行回答的图。虽然这用简单的Chain也能实现,但有助于我们理解Graph的构建模式。
fromtypingimportTypedDictfromlanggraph.graphimportStateGraph,ENDfromlangchain_openaiimportChatOpenAI# 1. 定义状态结构classGraphState(TypedDict):question:stranswer:str# 2. 初始化LLMllm=ChatOpenAI(model="gpt-4o-mini")# 3. 定义节点函数defgenerate_answer(state:GraphState):"""节点:调用LLM生成答案"""response=llm.invoke(state["question"])return{"answer":response.content}# 4. 构建图builder=StateGraph(GraphState)# 添加节点builder.add_node("generate",generate_answer)# 设置入口点builder.set_entry_point("generate")# 设置结束点(从`generate`节点指向预定义的`END`)builder.add_edge("generate",END)# 5. 编译图graph=builder.compile()# 6. 运行图initial_state={"question":"LangGraph是什么?"}result=graph.invoke(initial_state)print(result["answer"])这个图虽然简单,但已经包含了所有核心要素:定义状态、创建节点、构建边、编译和执行。
进阶实战:构建一个具备工具调用能力的智能体
真正的威力在于处理复杂逻辑。下面我们构建一个更实用的智能体,它可以根据用户问题决定是直接回答,还是需要调用一个搜索工具。
fromtypingimportTypedDict,Literalfromlanggraph.graphimportStateGraph,ENDfromlangchain_openaiimportChatOpenAIfromlangchain_community.toolsimportDuckDuckGoSearchRunfromlangchain_core.messagesimportHumanMessage,SystemMessage# 定义更丰富的状态classAgentState(TypedDict):messages:list# 对话历史question:strneeds_search:bool# 是否需要搜索search_result:strfinal_answer:str# 初始化组件llm=ChatOpenAI(model="gpt-4o-mini")search_tool=DuckDuckGoSearchRun()# 节点1:路由决策defroute_question(state:AgentState):"""判断问题是否需要联网搜索"""system_prompt="你是一个路由助手。判断用户的问题是否需要实时信息或网络搜索才能准确回答。如果需要,回复'search',否则回复'answer'。"prompt=f"用户问题:{state['question']}"response=llm.invoke([SystemMessage(content=system_prompt),HumanMessage(content=prompt)])decision=response.content.strip().lower()return{"needs_search":decision=="search"}# 节点2:调用搜索工具defcall_search_tool(state:AgentState):"""执行网络搜索"""result=search_tool.run(state["question"])return{"search_result":result}# 节点3:基于知识回答defanswer_with_knowledge(state:AgentState):"""仅用模型已有知识回答"""response=llm.invoke(f"请回答以下问题:{state['question']}")return{"final_answer":response.content}# 节点4:综合搜索结果的答案defanswer_with_search(state:AgentState):"""结合搜索结果进行回答"""prompt=f""" 用户问题:{state['question']}搜索到的相关信息:{state['search_result']}请根据以上信息,给出一个准确、全面的回答。 """response=llm.invoke(prompt)return{"final_answer":response.content}# 构建图builder=StateGraph(AgentState)# 添加所有节点builder.add_node("route",route_question)builder.add_node("search",call_search_tool)builder.add_node("answer_directly",answer_with_knowledge)builder.add_node("answer_with_search",answer_with_search)# 设置入口点builder.set_entry_point("route")# 添加条件边:根据路由结果决定走向defdecide_next_node(state:AgentState)->Literal["search","answer_directly"]:ifstate["needs_search"]:return"search"else:return"answer_directly"builder.add_conditional_edges("route",decide_next_node,{"search":"search","answer_directly":"answer_directly"})# 添加普通边builder.add_edge("search","answer_with_search")builder.add_edge("answer_directly",END)builder.add_edge("answer_with_search",END)# 编译图agent_graph=builder.compile()# 运行智能体result1=agent_graph.invoke({"question":"Python中如何读写JSON文件?","messages":[]})print("【无需搜索的问题】")print(f"答案:{result1['final_answer'][:200]}...")result2=agent_graph.invoke({"question":"今天北京天气怎么样?","messages":[]})print("\n【需要搜索的问题】")print(f"答案:{result2['final_answer'][:200]}...")这个示例展示了LangGraph如何优雅地处理条件路由和线性序列的组合。route节点后的条件边实现了智能决策,而search到answer_with_search的边则是一个简单的顺序流。
可视化与调试:graph.get_graph().draw_mermaid()
LangGraph一个非常强大的特性是可视化。你可以轻松地将你构建的图导出为Mermaid图表,直观地理解工作流。
# 接上例,可视化智能体图try:# 绘制并显示Mermaid图代码mermaid_code=agent_graph.get_graph().draw_mermaid()print("将以下代码复制到Mermaid编辑器中查看(如mermaid.live):")print(mermaid_code)exceptExceptionase:print(f"生成Mermaid代码时出错:{e}")生成的Mermaid图可以清晰地展示节点、边和决策逻辑,是设计和调试复杂工作流的利器。
最佳实践与注意事项
- 状态设计要精简:只将需要在节点间传递的数据放入
State。避免状态过于庞大,影响性能和清晰度。 - 节点职责单一:每个节点应只完成一件明确的事情,遵循单一职责原则,使图更易于维护和测试。
- 善用条件边与循环:对于需要多轮对话(如直到满足某个条件)的智能体,可以利用条件边指向自身或某个判断节点,实现
while循环。 - 错误处理:考虑在图中添加专门的错误处理节点,或者使用
try-catch包装可能失败的节点逻辑,确保图的健壮性。 - 持久化状态:对于长会话应用,需要将
State持久化到数据库。LangGraph通常与State的checkpointer结合使用,来实现状态的保存与恢复。