LangGraph - 构建状态化 AI 工作流的框架
什么是 LangGraph LangGraph 是 LangChain 生态系统中的一个框架,专门用于构建有状态的、多步骤的 AI 应用程序。它通过图(Graph)的方式来定义和管理复杂的 AI 工作流,让开发者能够创建具有循环、条件分支和持久化状态的智能体系统。 LangGraph 的核心理念是将 AI 应用程序建模为状态机,其中节点代表操作步骤,边代表流程控制,状态在节点之间传递和更新。 核心概念 State(状态) State 是在整个工作流中传递和更新的数据结构。通常使用 TypedDict 定义: from typing import TypedDict, Annotated from langgraph.graph import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] user_input: str analysis_result: str Node(节点) Node 是执行具体操作的函数,接收当前状态并返回更新后的状态: def research_node(state: AgentState): # 执行研究任务 result = do_research(state['user_input']) return { "analysis_result": result, "messages": [("assistant", f"研究完成:{result}")] } Edge(边) Edge 定义节点之间的连接关系: Normal Edge(普通边):直接连接两个节点 Conditional Edge(条件边):根据状态决定下一个节点 from langgraph.graph import StateGraph, END # 普通边 workflow.add_edge("node_a", "node_b") # 条件边 def should_continue(state): if state['done']: return END return "continue_node" workflow.add_conditional_edges( "decision_node", should_continue ) Graph(图) Graph 是完整的工作流定义,包含所有节点、边和状态管理: from langgraph.graph import StateGraph workflow = StateGraph(AgentState) # 添加节点 workflow.add_node("research", research_node) workflow.add_node("analyze", analyze_node) workflow.add_node("respond", respond_node) # 定义边 workflow.add_edge("research", "analyze") workflow.add_edge("analyze", "respond") # 设置入口点 workflow.set_entry_point("research") # 编译图 app = workflow.compile() 核心特性 1. 状态持久化 LangGraph 支持状态的持久化存储,可以实现: ...