LangGraph：从原型到生产的Agent框架 LangGraph最大的优势不在于功能丰富，而在于它对生产环境的认真对待——状态管理、检查点、人机协作、错误处理，这些生产级需求被设计在框架核心而非附加功能。
状态管理 定义Agent状态 from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated, List import operator class AgentState(TypedDict): messages: Annotated[List, operator.add] # 消息列表（追加） current_task: str # 当前任务 completed_steps: List[str] # 已完成步骤 tool_results: dict # 工具结果 error_count: int # 错误计数 human_feedback: str # 人类反馈 next_action: str # 下一步行动 # 创建图 graph = StateGraph(AgentState) 状态更新模式 def research_node(state: AgentState): """研究节点：执行信息检索""" query = state["current_task"] results = search_tool(query) # 状态更新（自动合并） return { "messages": [{"role": "assistant", "content": f"找到{len(results)}条结果"}], "tool_results": {"search": results}, "completed_steps": state["completed_steps"] + ["research"], "next_action": "analyze" } def analyze_node(state: AgentState): """分析节点：分析检索结果""" results = state["tool_results"]["search"] analysis = llm.analyze(results) return { "messages": [{"role": "assistant", "content": analysis}], "completed_steps": state["completed_steps"] + ["analyze"], "next_action": "write" if analysis else "research" # 分析不足则重新检索 } 检查点与恢复 持久化执行状态 from langgraph.checkpoint import MemorySaver, SqliteSaver # 使用SQLite持久化 checkpointer = SqliteSaver.from_conn_string("agent.db") graph = StateGraph(AgentState) graph.add_node("research", research_node) graph.add_node("analyze", analyze_node) graph.add_node("write", write_node) graph.add_edge("research", "analyze") graph.add_conditional_edges("analyze", lambda s: s["next_action"]) graph.add_edge("write", END) app = graph.compile(checkpointer=checkpointer) # 执行（可以中断和恢复） config = {"configurable": {"thread_id": "task-123"}} result = app.invoke( {"current_task": "分析AI芯片市场", "messages": []}, config=config ) # 恢复执行 restored = app.get_state(config) # 可以从任意检查点恢复 检查点策略 class CheckpointStrategy: def __init__(self): self.saver = SqliteSaver.from_conn_string("checkpoints.db") def should_checkpoint(self, state): """决定是否需要检查点""" # 关键步骤后检查 if state.get("completed_steps"): last_step = state["completed_steps"][-1] if last_step in ["research", "analyze", "write"]: return True # 错误后检查 if state.get("error_count", 0) > 0: return True return False 人机协作 人工审批节点 # 在关键步骤前暂停，等待人工确认 app = graph.compile( checkpointer=checkpointer, interrupt_before=["publish"] # 发布前暂停 ) # 执行到publish节点前会暂停 result = app.invoke( {"current_task": "撰写技术报告"}, config={"configurable": {"thread_id": "task-456"}} ) # 人工审查后继续 if human_approved: result = app.invoke(None, config=config) # 传入None继续执行 else: # 人工提供修改意见 result = app.invoke( {"human_feedback": "需要增加市场分析部分"}, config=config ) 交互式Agent def human_interaction_node(state: AgentState): """需要人工输入的节点""" # 展示当前状态 print(f"已完成步骤: {state['completed_steps']}") print(f"当前结果: {state.get('tool_results', {})}") # 请求人工输入 feedback = input("请提供反馈（直接回车确认）: ") return { "human_feedback": feedback, "next_action": "revise" if feedback else "continue" } 错误处理与重试 节点级错误处理 def robust_node(state: AgentState, max_retries=3): """带错误处理的节点""" try: result = execute_task(state["current_task"]) return { "tool_results": result, "error_count": 0, "next_action": "next" } except Exception as e: retry_count = state.get("error_count", 0) + 1 if retry_count < max_retries: # 重试 return { "error_count": retry_count, "next_action": "retry" # 重新执行当前节点 } else: # 超过重试次数，降级处理 return { "error_count": 0, "messages": [{"role": "system", "content": f"任务失败: {e}"}], "next_action": "fallback" } 条件边实现重试逻辑 graph.add_node("execute", robust_node) graph.add_node("fallback", fallback_node) # 正常流程 graph.add_edge("execute", "next_node") # 重试逻辑 graph.add_conditional_edges( "execute", lambda state: state.get("next_action"), { "retry": "execute", # 重试当前节点 "next": "next_node", # 正常进入下一步 "fallback": "fallback" # 降级处理 } ) 子图与模块化 # 将复杂Agent拆分为子图 def build_research_subgraph(): """研究子图""" subgraph = StateGraph(ResearchState) subgraph.add_node("search", search_node) subgraph.add_node("filter", filter_node) subgraph.add_node("summarize", summarize_node) subgraph.add_edge("search", "filter") subgraph.add_edge("filter", "summarize") subgraph.add_edge("summarize", END) return subgraph.compile() # 主图中嵌入子图 main_graph = StateGraph(AgentState) main_graph.add_node("research", build_research_subgraph()) # 嵌入子图 main_graph.add_node("write", write_node) main_graph.add_edge("research", "write") 并行执行 from langgraph.graph import StateGraph, END import operator from typing import Annotated class ParallelState(TypedDict): task: str results: Annotated[list, operator.add] # 并行结果追加 def parallel_research(state): """并行执行多个研究任务""" sub_tasks = decompose(state["task"]) # 并行执行 results = [] for sub_task in sub_tasks: result = research_agent.run(sub_task) results.append(result) return {"results": results} # 或者使用LangGraph的Send API实现真正的并行 from langgraph.constants import Send def fan_out(state): """扇出并行任务""" sub_tasks = decompose(state["task"]) return [ Send("research_node", {"sub_task": st}) for st in sub_tasks ] 生产部署 部署架构 class LangGraphDeployment: def __init__(self): self.config = { "runtime": { "framework": "FastAPI", "workers": 4, "timeout": 300, # 5分钟超时 }, "checkpoint": { "backend": "PostgreSQL", # 生产用PostgreSQL "cleanup_interval": 3600, # 1小时清理一次 "retention_days": 7, # 保留7天 }, "monitoring": { "trace_enabled": True, "metrics": ["latency", "success_rate", "token_usage"], "alerting": { "error_rate_threshold": 0.05, "latency_p99_threshold": 30000, # 30秒 } } } def deploy(self): # FastAPI服务 from fastapi import FastAPI app = FastAPI() @app.post("/agent/run") async def run_agent(task: str, thread_id: str): config = {"configurable": {"thread_id": thread_id}} result = await self.agent.ainvoke( {"current_task": task}, config=config ) return result return app 性能优化 class PerformanceOptimizer: def optimize_graph(self, graph): """图优化""" # 1. 节点合并：将总是顺序执行的节点合并 # 2. 冗余边移除：移除不会被执行的边 # 3. 缓存：对确定性节点启用缓存 optimized = graph # 启用缓存 for node in graph.nodes: if is_deterministic(node): node.enable_cache = True node.cache_ttl = 3600 return optimized 监控与可观测性 class AgentMonitor: def __init__(self): self.traces = [] def trace_execution(self, graph, input_state): """追踪Agent执行""" trace = { "input": input_state, "nodes_executed": [], "total_duration": 0, "token_usage": 0, "errors": [] } for node_name, node_output in graph.stream(input_state): trace["nodes_executed"].append({ "node": node_name, "duration": measure_duration(), "output": node_output, "timestamp": datetime.now() }) return trace def visualize(self, trace): """可视化执行轨迹""" return { "graph": render_execution_graph(trace), "timeline": render_timeline(trace), "bottlenecks": identify_bottlenecks(trace) } 结语 LangGraph的设计哲学是"为生产而构建"。它的图模型提供了精确的控制力，检查点机制保障了可靠性，人机协作支持了复杂业务流程。对于需要从原型走向生产的Agent系统，LangGraph是最稳妥的选择。学习曲线确实陡峭，但这是为生产级功能付出的合理代价——在生产环境中，可靠性和可控性远比开发便利性重要。
...
