LangGraph 完全实战指南

# 安装 pip install langgraph langchain-openai langchain-community # 设置 API Key export OPENAI_API_KEY=sk-xxx # 你的第一个 LangGraph Agent（复制以下代码到 demo.py 运行） from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, END from langgraph.prebuilt import ToolNode from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages from langchain_core.messages import BaseMessage class State(TypedDict): messages: Annotated[List[BaseMessage], add_messages] def get_weather(city: str) -> str: """获取城市天气""" return f'{{"city":"{city}","weather":"晴,22-28°C"}}' llm = ChatOpenAI(model="gpt-4o").bind_tools([get_weather]) def agent(s): return {"messages": [llm.invoke(s["messages"])]} def router(s): return "tools" if s["messages"][-1].tool_calls else END b = StateGraph(State) b.add_node("agent", agent); b.add_node("tools", ToolNode([get_weather])) b.set_entry_point("agent"); b.add_conditional_edges("agent", router); b.add_edge("tools", "agent") app = b.compile() print(app.invoke({"messages": [("human", "北京天气？")]}))

# 企业级 State 模板（分层设计） class EnterpriseState(TypedDict): # 对话层 messages: Annotated[List[BaseMessage], add_messages] session_id: str # 业务层 intent: Optional[str] entities: dict stage: str # greeting|collecting|processing|closing # 工具层 tool_calls: List[dict] tool_results: List[dict] error_count: int # 控制层 need_human: bool human_approved: Optional[bool] is_complete: bool # 监控层 total_tokens: Annotated[int, lambda c, n: c + n] events: Annotated[List[dict], lambda c, n: c + n]

# 7 种条件路由实战模式 # 1. LLM 工具决策 builder.add_conditional_edges("agent", lambda s: "tools" if s["messages"][-1].tool_calls else END) # 2. 金额阈值审批 builder.add_conditional_edges("audit", lambda s: "urgent" if s["amount"] > 100000 else "manager" if s["amount"] > 10000 else "auto") # 3. 对话阶段路由 builder.add_conditional_edges("analyze", lambda s: { "greeting": "collect", "collecting": "process", "confirming": "confirm", "closing": END }.get(s["stage"], "fallback")) # 4. 错误恢复路由 builder.add_conditional_edges("process", lambda s: "retry" if s["errors"] < 3 else "human") # 5. 动态分支 - 运行时决定 builder.add_conditional_edges("dispatcher", lambda s: s["next"]) # 6. 多条件组合 builder.add_conditional_edges("check", lambda s: [p for p in ["audit", "legal", "compliance"] if s[f"need_{p}"]] or [END]) # 7. 概率路由（A/B 测试） builder.add_conditional_edges("router", lambda _: "model_a" if random.random() < 0.5 else "model_b")

# 多工具 ReAct Agent 完整实现 import json, random from typing import TypedDict, Annotated, List from langchain_openai import ChatOpenAI from langchain_core.messages import BaseMessage from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langgraph.prebuilt import ToolNode class State(TypedDict): messages: Annotated[List[BaseMessage], add_messages] # ----- 工具定义 ----- def get_weather(city: str) -> str: """获取指定城市的实时天气""" db = {"北京":"晴 22-28°C","上海":"多云 25-30°C","广州":"阵雨 26-32°C","深圳":"多云 27-31°C"} return json.dumps({"city":city,"weather":db.get(city,"无数据"),"unit":"celsius"}) def calculate(expr: str) -> str: """计算数学表达式。expr: 如 '2+3*4'""" try: return json.dumps({"expr":expr,"result":eval(expr)}) except: return json.dumps({"error":"无法计算"}) def search_news(topic: str) -> str: """搜索最新新闻。topic: 搜索主题""" headlines = [f"AI 领域突破：{topic} 技术取得重大进展", f"2024 年 {topic} 市场规模增长 45%"] return json.dumps({"topic":topic,"news":headlines,"count":len(headlines)}) def send_email(to: str, subject: str, body: str) -> str: """发送邮件。to: 收件人, subject: 主题, body: 正文""" return json.dumps({"status":"sent","to":to,"subject":subject}) tools = [get_weather, calculate, search_news, send_email] llm = ChatOpenAI(model="gpt-4o", temperature=0).bind_tools(tools) def call_agent(s: State): return {"messages":[llm.invoke(s["messages"])]} def should_continue(s: State): last = s["messages"][-1] return "tools" if getattr(last,"tool_calls",None) else END b = StateGraph(State) b.add_node("agent", call_agent); b.add_node("tools", ToolNode(tools)) b.set_entry_point("agent"); b.add_conditional_edges("agent", should_continue) b.add_edge("tools", "agent") app = b.compile() # 使用示例：Agent 自主决定调用哪些工具、按什么顺序 for event in app.stream({"messages":[("human","帮我查查北京的天气，搜索AI的最新新闻，然后计算 256*8 等于多少，最后把结果发邮件给 admin@test.com")]}): for k,v in event.items(): print(f"[{k}]", v)

# 并行执行 + 汇聚 b.add_edge("dispatcher", ["research", "coding", "testing"]) b.add_edge(["research", "coding", "testing"], "merger") # 超时控制：并行节点最多等 30 秒 b.add_edge(["search_a", "search_b"], "merge", timeout=30)

# Map：对每份文档并行分析 docs = [...] # 10 份文档 for i in range(len(docs)): b.add_node(f"analyze_{i}", lambda s, d=docs[i]: analyze_doc(s, d)) b.add_edge("splitter", f"analyze_{i}") # Reduce：全部完成后汇总 b.add_edge([f"analyze_{i}" for i in range(len(docs))], "summarizer")

# 子图：独立的审批流程 sub = StateGraph(ApprovalState) sub.add_node("check", check_doc); sub.add_node("sign", digital_sign) sub.add_node("archive", archive_doc) sub.set_entry_point("check"); sub.add_edge("check", "sign") sub.add_edge("sign", "archive"); sub.add_edge("archive", END) approval_graph = sub.compile() # 主图中嵌入子图 main.add_node("approval", approval_graph) # 子图内部完全封装，主图无需关心实现细节

# 一票通过的完整实现 builder.add_node("auto_gen", generate_content) builder.add_node("human_review", lambda s: s) # 占位节点 builder.add_node("publish", publish_content) builder.add_edge("auto_gen", "human_review") builder.add_conditional_edges("human_review", lambda s: "publish" if s["approved"] else "auto_gen") builder.add_edge("publish", END) # 在 human_review 前暂停 app = builder.compile(interrupt_before=["human_review"]) # 执行流：auto_gen → 【暂停】→ 人工审批 → 继续 config = {"configurable": {"thread_id": "doc_001"}} # 第一步：自动生成内容 for event in app.stream(inputs, config): print(event) # 人工审核通过后：更新状态并恢复执行 app.update_state(config, {"approved": True}) app.invoke(None, config) # 继续到 publish 节点

# PostgresSaver 生产配置 pip install langgraph-checkpoint-postgres from langgraph.checkpoint.postgres import PostgresSaver conn_string = "postgresql://user:pass@host:5432/langgraph" memory = PostgresSaver.from_conn_string(conn_string) memory.setup() # 建表 app = builder.compile(checkpointer=memory) # 跨会话记忆：同一 thread_id 共享状态 c1 = {"configurable": {"thread_id": "user_abc"}} app.invoke({"messages": [("human", "我叫小明")]}, c1) app.invoke({"messages": [("human", "我叫什么？")]}, c1) # Agent 回答: 你叫小明 ✨

# 监督模式：主管 + 专家团队 class TeamState(TypedDict): task: str; research: str; code: str; review: str; final: str def supervisor(s: TeamState): # 分析任务并分配给专家 if "调研" in s["task"]: return {"research": "进行中"} elif "编码" in s["task"]: return {"code": "进行中"} else: return {"review": "进行中"} b = StateGraph(TeamState) b.add_node("supervisor", supervisor) b.add_node("research_agent", research_agent_subgraph) b.add_node("coding_agent", coding_agent_subgraph) b.add_node("review_agent", review_agent_subgraph) b.set_entry_point("supervisor") b.add_conditional_edges("supervisor", lambda s: "research_agent" if s["research"] else "coding_agent" if s["code"] else "review_agent") b.add_edge(["research_agent", "coding_agent", "review_agent"], "supervisor")

# 输入过滤节点示例 def sanitize_input(state): user_msg = state["messages"][-1].content # 检测 prompt 注入模式 injections = ["ignore previous", "forget all", "system prompt"] if any(w in user_msg.lower() for w in injections): return {"messages": [("ai", "抱歉，检测到异常输入，已终止处理。")], "blocked": True} return {"blocked": False} # 在入口处添加过滤节点 b.add_node("sanitizer", sanitize_input) b.set_entry_point("sanitizer") b.add_conditional_edges("sanitizer", lambda s: END if s["blocked"] else "agent")

# Pytest 测试示例 def test_call_agent(): state = {"messages": [("human", "hello")]} result = call_agent(state) assert "messages" in result assert len(result["messages"]) == 1 def test_graph_end_to_end(): app = builder.compile() result = app.invoke({"messages": [("human", "北京天气")]}) assert result["messages"][-1].content assert "北京" in result["messages"][-1].content

# FastAPI + Docker 部署模板 from fastapi import FastAPI, HTTPException from pydantic import BaseModel import uvicorn api = FastAPI(title="LangGraph Agent API") class ChatReq(BaseModel): message: str thread_id: str = "default" user_id: str = "anonymous" class ChatRes(BaseModel): reply: str thread_id: str tokens: int @api.post("/chat", response_model=ChatRes) async def chat(req: ChatReq): config = {"configurable": {"thread_id": req.thread_id}} try: result = await app.ainvoke( {"messages": [("human", req.message)]}, config) reply = result["messages"][-1].content return ChatRes(reply=reply, thread_id=req.thread_id, tokens=result.get("total_tokens", 0)) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) # 健康检查 @api.get("/health") async def health(): return {"status": "ok"} if __name__ == "__main__": uvicorn.run(api, host="0.0.0.0", port=8000)