1. 项目概述从“一团乱麻”到“清晰蓝图”如果你最近在折腾AI Agent尤其是尝试把多个Agent串起来干活大概率经历过这种痛苦写了几百上千行的Prompt试图用自然语言描述清楚每个Agent的职责、它们之间如何对话、在什么条件下该做什么事。结果呢代码逻辑像意大利面条一样纠缠不清Agent们时不时“精神错乱”要么卡在某个循环里出不来要么在错误的时间说了不该说的话。调试起来更是噩梦你根本分不清是Prompt写得不够好还是逻辑本身就有漏洞。这正是标题里说的“把Agent写成一团Prompt”的典型困境。问题的根源在于我们试图用线性的、描述性的Prompt去管理一个本质上非线性的、充满状态跳转的复杂系统。多Agent协作不是简单的“A说完B说”它更像一个精密的流水线或一个决策网络每个节点Agent根据当前系统的整体状态比如用户输入的历史、中间结果、外部API的反馈来决定自己是否激活、以及激活后做什么。这种“根据状态决定行为”的模式在计算机科学里有一个非常成熟的理论模型来对应——状态机State Machine。而LangGraph正是LangChain官方推出的、专门用来解决这个问题的框架。它不是一个新的大模型而是一个编排Orchestration工具。它的核心思想就是让你用定义状态机的方式来定义你的多Agent系统。你把每个Agent看作状态机里的一个“节点”Node把Agent之间的协作逻辑和流转条件定义为连接节点的“边”Edge。这样一来整个系统的运行逻辑就从一团模糊的Prompt描述变成了一张清晰可见、可调试、可控制的“流程图”。简单来说这个项目的核心价值是将多Agent系统的开发从“Prompt工程”的玄学领域拉回到“软件工程”的可控领域。你不再需要和晦涩的Prompt较劲而是像设计一个普通程序一样用代码定义状态、节点和流转让系统的行为变得可预测、可维护。这对于构建真正可靠、可用于生产环境的AI应用至关重要。2. 核心设计用状态机思维重构Agent协作在深入代码之前我们必须先彻底理解背后的设计哲学。为什么状态机是更优解我们通过一个具体的场景来对比。假设我们要构建一个“旅行规划助手”它包含三个Agent需求理解Agent分析用户模糊的需求如“我想去个暖和的地方放松一下”。信息查询Agent调用外部API获取目的地天气、机票、酒店信息。方案生成Agent综合信息生成一份详细的旅行计划报告。2.1 传统Prompt编排的典型困局如果用传统方式你可能会写一个超级长的Prompt给一个“主Agent”或者写多个Prompt让它们互相调用。伪代码逻辑可能长这样# 伪代码示意混乱的流程 user_input “我想去个暖和的地方放松一下” context {} # 尝试用条件判断控制流程 if “需求不清晰” in 需求理解Agent(user_input): # 让需求理解Agent去追问用户 follow_up_question 需求理解Agent.generate_question() # 但如何把追问和后续流程衔接状态保存在哪 context[“waiting_for_clarification”] True # ... 逻辑开始变得复杂和脆弱 elif “需要查询信息” in 需求理解Agent(user_input): destinations 需求理解Agent.extract_destination() for dest in destinations: weather 信息查询Agent.query_weather(dest) # 如果查询失败怎么办是重试、跳过还是报错 if weather is None: # 错误处理逻辑侵入主流程 context[“error”] f“无法获取{dest}天气” # 该由哪个Agent来处理这个错误 # 信息收集齐了再调用方案生成Agent report 方案生成Agent(context)你会发现控制逻辑if-else、业务逻辑调用Agent和状态管理context字典全部耦合在一起。增加一个“预算检查Agent”或者修改某个环节的判断条件都可能牵一发而动全身。这团“Prompt”或者说混杂着Prompt的逻辑代码极其难以维护和调试。2.2 LangGraph的状态机模型解析LangGraph引入了清晰的分层概念状态State这是一个共享的、类型化的数据容器。它定义了在整个流程中有哪些数据需要被传递和修改。比如我们的旅行规划状态可以定义为from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages import operator class State(TypedDict): # 消息历史LangGraph内置支持 messages: Annotated[List[str], add_messages] # 用户原始输入 user_request: str # 理解后的结构化需求 parsed_requirements: dict # 查询到的目的地列表 candidate_destinations: List[str] # 收集到的目的地详细信息 destination_details: List[dict] # 最终生成的报告 travel_plan: str # 用于控制流程的标志位例如是否需用户澄清 needs_clarification: bool clarification_question: str这个State就像整个系统的“记忆白板”所有Agent都读写这个白板上的特定部分。这解决了状态分散在各处的问题。节点Node每个节点是一个函数通常封装了一个Agent或一个工具调用。它接收当前的State执行操作然后返回更新后的State。def requirement_agent_node(state: State) - State: “”“需求理解节点”“” # 1. 从state中读取输入 user_input state[“user_request”] # 2. 调用实际的LLM Agent这里用LCEL链示意 llm_chain create_requirement_chain() # 这里封装了Prompt和LLM调用 result llm_chain.invoke({“input”: user_input}) # 3. 解析结果更新State state[“parsed_requirements”] result[“parsed”] state[“needs_clarification”] result[“needs_clarification”] state[“clarification_question”] result[“question”] if result[“needs_clarification”] else “” # 4. 返回更新后的State return state每个节点职责单一只关心自己的输入输出和State的更新。边Edge边决定了流程的走向。在LangGraph中这通过条件判断函数conditional_edge或路由器Router来实现。这是将“一团Prompt”逻辑转化为“清晰路径”的关键。from langgraph.graph import END, StateGraph # 构建图 workflow StateGraph(State) # 添加节点 workflow.add_node(“understand_requirements”, requirement_agent_node) workflow.add_node(“query_information”, query_agent_node) workflow.add_node(“generate_plan”, plan_agent_node) workflow.add_node(“ask_user”, ask_user_node) # 设置入口点 workflow.set_entry_point(“understand_requirements”) # 添加边从“需求理解”节点出发根据状态决定下一步 def route_after_understanding(state: State) - str: if state[“needs_clarification”]: return “ask_user” # 需要澄清跳转到询问用户节点 elif state[“parsed_requirements”].get(“destination_clear”): return “query_information” # 目的地明确跳转到查询节点 else: return “generate_plan” # 无需查询例如用户只要概念建议直接生成计划 workflow.add_conditional_edges( “understand_requirements”, route_after_understanding, { “ask_user”: “ask_user”, “query_information”: “query_information”, “generate_plan”: “generate_plan”, } ) # 添加其他边... workflow.add_edge(“query_information”, “generate_plan”) workflow.add_edge(“ask_user”, “understand_requirements”) # 用户回答后重新理解需求 workflow.add_edge(“generate_plan”, END) # 最终节点指向结束 # 编译图 app workflow.compile()通过add_conditional_edges我们清晰地声明了“在需求理解节点之后如果needs_clarification为真就去问用户否则如果目的地明确就去查信息否则直接生成计划”。这个逻辑是声明式的、集中管理的与节点内部的Agent实现完全解耦。这种设计的巨大优势在于可视化LangGraph可以生成系统的流程图一目了然。可调试你可以跟踪State在每个节点的变化精准定位问题出在哪个环节。可维护修改流程只需增删节点或调整边的条件不会影响其他部分。支持复杂拓扑轻松实现循环比如多次澄清、并行同时查询多个目的地、分支与合并等复杂流程。关键心得在LangGraph中设计系统时要花最多的时间在定义State的结构和Edge的条件上。State是你的数据模型Edge是你的业务逻辑。把它们设计好了节点里的Agent实现反而可以相对独立和简单。这和我们设计数据库Schema和API接口的思路是相通的。3. 实操构建一个可运行的多Agent问答系统理论说再多不如动手。我们来构建一个相对完整但核心清晰的多Agent问答系统。这个系统包含两个Agent一个“检索Agent”负责从知识库找资料一个“回答Agent”负责组织答案。如果检索不到资料系统会主动告知用户“我不知道”。3.1 环境准备与状态定义首先安装必要库并定义状态。pip install langgraph langchain-openai langchain-chroma# 导入 from typing import TypedDict, Annotated, List, Optional from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages import operator from langchain_openai import ChatOpenAI from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings from langchain.schema import Document from langchain_core.prompts import ChatPromptTemplate # 1. 定义状态State class AgentState(TypedDict): “”“整个多Agent系统的共享状态。”“” # 消息历史使用LangGraph提供的注解自动处理追加逻辑 messages: Annotated[List[str], add_messages] # 用户当前的问题 question: str # 检索Agent找到的相关文档 retrieved_docs: Optional[List[Document]] # 最终给用户的答案 final_answer: Optional[str] # 一个标志位控制流程 should_respond: bool这里的关键是Annotated[List[str], add_messages]。add_messages是一个归约器reducer它定义了当多个节点都想更新messages字段时如何合并这些更新这里是追加。这是LangGraph管理共享状态的精髓之一。3.2 实现核心节点函数接下来我们实现两个Agent节点和一个路由判断逻辑。# 初始化LLM和向量数据库这里用内存示例 llm ChatOpenAI(model“gpt-4o-mini”, temperature0) embeddings OpenAIEmbeddings() # 假设我们有一个已存在的知识库向量存储 vectorstore Chroma(embedding_functionembeddings, persist_directory“./chroma_db”) retriever vectorstore.as_retriever(search_kwargs{“k”: 3}) # 2. 实现检索节点 def retrieve_node(state: AgentState) - AgentState: “”“检索节点从知识库中查找与问题相关的文档。”“” print(f“【检索节点】正在处理问题{state[‘question’]}”) try: docs retriever.invoke(state[“question”]) state[“retrieved_docs”] docs print(f“【检索节点】检索到 {len(docs)} 条相关文档。”) # 如果没找到任何相关文档我们设置一个标志 if not docs: state[“should_respond”] False # 触发“无法回答”流程 else: state[“should_respond”] True # 正常进入回答流程 except Exception as e: print(f“【检索节点】检索过程出错{e}”) state[“retrieved_docs”] [] state[“should_respond”] False return state # 3. 实现回答生成节点 def generate_answer_node(state: AgentState) - AgentState: “”“回答生成节点基于检索到的文档生成答案。”“” print(“【回答节点】正在生成答案...”) docs state[“retrieved_docs”] context “\n\n”.join([doc.page_content for doc in docs]) # 构建Prompt明确要求基于上下文回答 prompt ChatPromptTemplate.from_messages([ (“system”, “你是一个专业的助手请严格根据以下提供的信息来回答问题。如果信息不足以回答问题请明确说‘根据已有信息无法回答’。\n\n相关信息\n{context}”), (“human”, “{question}”) ]) chain prompt | llm response chain.invoke({“context”: context, “question”: state[“question”]}) state[“final_answer”] response.content # 将最终答案也放入消息历史方便后续查看 state[“messages”].append(response.content) print(f“【回答节点】答案生成完毕。”) return state # 4. 实现“无法回答”节点 def cannot_answer_node(state: AgentState) - AgentState: “”“无法回答节点当检索不到相关信息时调用。”“” print(“【无法回答节点】被触发。”) state[“final_answer”] “抱歉根据我现有的知识库无法回答这个问题。请尝试换一种问法或询问其他问题。” state[“messages”].append(state[“final_answer”]) return state # 5. 实现路由判断逻辑 def route_after_retrieve(state: AgentState) - str: “”“在检索节点之后决定下一步是生成答案还是直接告知无法回答。”“” # 根据检索节点设置的 should_respond 标志来决定 if state.get(“should_respond”, False): return “generate_answer” else: return “cannot_answer”3.3 组装成图并运行现在我们把节点和边组装成一个完整的、可执行的工作流。# 6. 创建状态图并构建工作流 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(“retrieve”, retrieve_node) workflow.add_node(“generate_answer”, generate_answer_node) workflow.add_node(“cannot_answer”, cannot_answer_node) # 设置入口点所有对话都从检索开始 workflow.set_entry_point(“retrieve”) # 添加条件边检索完成后根据路由函数决定去向 workflow.add_conditional_edges( “retrieve”, route_after_retrieve, { “generate_answer”: “generate_answer”, “cannot_answer”: “cannot_answer” } ) # 添加普通边回答生成节点和无法回答节点都指向结束 workflow.add_edge(“generate_answer”, END) workflow.add_edge(“cannot_answer”, END) # 编译图得到可执行的应用 app workflow.compile() # 7. 运行测试 # 初始化状态 initial_state: AgentState { “messages”: [], # 消息历史初始为空 “question”: “LangGraph的主要用途是什么”, “retrieved_docs”: None, “final_answer”: None, “should_respond”: True } print(“开始执行工作流...\n”) # 运行图 final_state app.invoke(initial_state) print(“\n工作流执行完毕。”) print(“-” * 30) print(f“用户问题{final_state[‘question’]}”) print(f“最终答案{final_state[‘final_answer’]}”) print(f“消息历史{final_state[‘messages’]}”)当你运行这段代码时控制台会清晰地打印出每个节点的执行日志开始执行工作流... 【检索节点】正在处理问题LangGraph的主要用途是什么 【检索节点】检索到 3 条相关文档。 【回答节点】正在生成答案... 【回答节点】答案生成完毕。 工作流执行完毕。 ------------------------------ 用户问题LangGraph的主要用途是什么 最终答案LangGraph是LangChain框架中的一个库主要用于构建复杂的、有状态的、多参与者的AI应用。它通过图Graph的结构来编排和协调多个LLM调用、工具使用以及自定义函数特别擅长处理带有循环、分支和状态管理的对话或工作流...这个简单的例子已经展示了LangGraph的核心价值流程可视化、状态可控、逻辑清晰。你可以通过app.get_graph().draw_mermaid()生成Mermaid流程图直观地看到retrieve - (generate_answer | cannot_answer) - END的结构。实操心得在定义State时务必提前规划好哪些数据是“只读”的如初始问题哪些是“可写”的如检索结果、最终答案。清晰的State设计是构建稳定工作流的基础。另外为每个节点函数添加清晰的日志打印对于调试复杂工作流有奇效。4. 高级模式与生产级考量掌握了基础构建后我们可以探索更强大的模式让系统更健壮、更高效。4.1 并行执行与子图提升效率与模块化我们的旅行规划例子中查询多个目的地信息是可以并行进行的。LangGraph支持通过Pregel其底层运行时实现节点的并行执行。from langgraph.graph import StateGraph from typing import List import asyncio class ParallelState(TypedDict): destinations: List[str] weather_results: List[Optional[float]] hotel_prices: List[Optional[float]] def fetch_weather(destination: str) - float: # 模拟天气查询API time.sleep(0.5) return 25.0 # 模拟返回值 def fetch_hotel_price(destination: str) - float: # 模拟酒店查询API time.sleep(0.8) return 500.0 # 并行节点同时获取天气和酒店价格 async def parallel_fetch_node(state: ParallelState): destinations state[“destinations”] # 使用asyncio.gather并行执行IO密集型任务 weather_tasks [asyncio.to_thread(fetch_weather, d) for d in destinations] hotel_tasks [asyncio.to_thread(fetch_hotel_price, d) for d in destinations] state[“weather_results”] await asyncio.gather(*weather_tasks) state[“hotel_prices”] await asyncio.gather(*hotel_tasks) return state对于更复杂的、可复用的功能单元可以将其封装成子图Subgraph。例如将“目的地信息收集”包含并行查询天气、酒店、景点封装成一个子图在主图中作为一个节点调用。这极大提升了代码的模块化和复用性。from langgraph.graph import StateGraph as SubStateGraph # 1. 定义子图的状态可能只是主状态的一部分 class DestinationInfoState(TypedDict): destination: str weather: Optional[float] hotel_price: Optional[float] attractions: List[str] # 2. 构建子图 sub_graph_builder SubStateGraph(DestinationInfoState) # ... 在子图中添加获取天气、价格、景点的节点和边 ... sub_graph sub_graph_builder.compile() # 3. 在主图中可以将子图作为一个“超级节点”添加 main_workflow.add_node(“fetch_destination_info”, sub_graph)4.2 持久化与检查点实现长时记忆与中断恢复生产系统中的Agent往往需要处理长时间运行的、可能中断的会话。LangGraph的检查点Checkpointing机制可以自动保存每个步骤后的完整状态允许工作流从任意历史步骤恢复。from langgraph.checkpoint.aiosqlite import AsyncSqliteSaver import asyncio async def main(): # 使用SQLite持久化检查点 memory AsyncSqliteSaver.from_conn_string(“:memory:”) # 生产环境用文件路径 workflow StateGraph(AgentState) # ... 添加节点和边 ... # 编译时传入检查点存储器 app workflow.compile(checkpointermemory) # 配置线程支持中断和恢复 config {“configurable”: {“thread_id”: “user_123_session_1”}} # 第一次执行 initial_state {“messages”: [], “question”: “第一个问题”} result1 await app.ainvoke(initial_state, configconfig) print(f“第一次执行后状态ID: {result1[‘__langgraph_node__’]}”) # 会有一个节点ID # 模拟系统中断... # 第二次基于同一个thread_id恢复执行可以传入新的输入 # LangGraph会从上次的检查点继续 new_state_snapshot {“messages”: result1[“messages”], “question”: “基于刚才的回答我的第二个问题是...”} result2 await app.ainvoke(new_state_snapshot, configconfig) # 会从上次结束的节点之后开始这对于构建需要记忆上下文、支持用户中途离开再回来的聊天机器人或复杂工作流至关重要。4.3 错误处理与人工干预在复杂流程中错误不可避免。LangGraph允许你定义错误处理节点。from langgraph.graph import StateGraph, START, END from langgraph.types import Command, interrupt def action_node(state): # 某个可能失败的操作 if something_went_wrong: # 不是返回state而是返回一个“命令”要求转移到错误处理节点 return Command(goto“handle_error”, update{“error_info”: “具体错误信息”}) state[“result”] “成功” return state def handle_error_node(state): print(f“处理错误{state.get(‘error_info’)}”) # 可以尝试修复、重试或者将错误信息记录到state由后续节点决定 state[“error_handled”] True return state workflow StateGraph(…) workflow.add_node(“action”, action_node) workflow.add_node(“handle_error”, handle_error_node) workflow.add_edge(START, “action”) # 注意action节点通过Command动态跳转所以这里不预先定义从action出发的边 workflow.add_edge(“handle_error”, END) # 错误处理后结束或去其他节点更高级的场景是人工干预Human-in-the-loop。当Agent不确定时可以暂停工作流将决策权交给人。def agent_node(state): if confidence threshold: # 触发中断等待人工输入 raise interrupt(“需要人工确认下一步操作”) # ... 正常处理 # 在外部你可以检查工作流状态如果处于中断则注入人工决策 if app.get_state(config).values.get(“__langgraph_interrupt__”): human_decision input(“请决定下一步A/B/C”) # 将人工决策作为更新注入继续工作流 app.update_state(config, {“human_feedback”: human_decision})生产环境建议对于关键业务流务必实现完善的错误处理和状态持久化。检查点机制不仅能用于恢复还能用于审计和调试你可以回放任意一次会话的完整状态变迁。同时为耗时长的节点设置超时并在图中设计“超时处理”分支是保证系统可用性的关键。5. 避坑指南与效能优化在实际项目中踩过一些坑后我总结出以下经验能帮你节省大量调试时间。5.1 状态设计中的常见陷阱状态字段过多过杂不要把所有数据都塞进一个State。应该按功能模块划分或者使用嵌套的TypedDict。否则State会变得难以理解且每次更新都可能触发不必要的序列化/反序列化开销。坏实践State包含user_input, parsed_data, api_result_1, api_result_2, temp_variable, final_output, debug_log...好实践class SubStateA(TypedDict): data: str result: dict class MainState(TypedDict): input: str module_a: SubStateA module_b: dict output: str忽略了状态的不可变性LangGraph的节点函数应该返回一个新的状态字典而不是修改传入的状态。虽然Python中直接修改传入的dict可能暂时可行但这违背了函数式编程的原则在并发或持久化时可能导致难以预料的问题。正确做法在节点函数内先创建状态的副本或更新字典再返回。def good_node(state: State) - State: new_state state.copy() # 或使用 {**state} 展开 new_state[“processed_data”] process(state[“raw_data”]) return new_state5.2 流程控制与循环的注意事项避免无限循环这是状态机最常见的坑。例如Agent A产生输出给Agent BAgent B又产生输出给Agent A如果没有终止条件就会死循环。务必在条件边conditional_edge中设置明确的、最终能走向END的路径。可以为循环设置最大迭代次数并将其作为State的一部分。class StateWithLoop(TypedDict): messages: List[str] iteration_count: int 0 max_iterations: int 5 def should_continue(state: StateWithLoop) - str: state[“iteration_count”] 1 if state[“iteration_count”] state[“max_iterations”]: return “force_end” elif some_condition(state): return “continue_loop” else: return “end_normally”理解add_messages归约器add_messages会自动将节点返回的messages列表与原有列表合并追加。如果你某个节点的目的是“替换”或“清空”消息历史就不能直接使用这个注解。你需要定义自己的归约器或者将该字段从add_messages注解中移除改为手动管理。# 自定义归约器示例只保留最新的N条消息 def keep_last_n(old_messages: List, new_messages: List) - List: all_messages old_messages new_messages return all_messages[-10:] # 只保留最后10条 class MyState(TypedDict): messages: Annotated[List[str], keep_last_n] # 使用自定义归约器5.3 性能与调试技巧节点函数应保持纯净节点函数最好只包含业务逻辑避免副作用。将API调用、数据库查询等IO操作封装在节点内部是合理的但要做好异常处理。避免在节点内修改全局变量或外部文件这会影响工作流的可重现性和可调试性。利用可视化工具在开发初期多用app.get_graph().draw_mermaid()生成流程图。一张图能帮你快速发现逻辑设计上的疏漏比如某个节点没有出口成了死胡同或者形成了意外的循环。启用详细日志LangGraph Pregel运行时提供了详细的执行日志。通过设置环境变量LANGGRAPH_LOG“verbose”或在代码中配置你可以看到每个节点的输入、输出、耗时这对于性能分析和问题定位至关重要。对耗时节点进行超时设置对于调用外部API或执行复杂计算的节点在节点函数内部使用asyncio.wait_for或设置超时防止单个节点卡住整个工作流。并在图中设计超时处理分支。测试策略不要只测试整个工作流的端到端流程。要针对每个节点函数编写单元测试模拟不同的输入State验证其输出。然后测试各个条件边路由函数的逻辑。最后再进行集成测试。这种分层的测试策略能极大提升开发效率。从“一团Prompt”到“可控状态机”的转变不仅仅是换了一个工具更是思维模式的升级。LangGraph迫使你以更工程化、更结构化的方式思考AI Agent的协作问题。当你习惯了这种模式后你会发现构建复杂、可靠的智能体系统不再是一件令人畏惧的事情。你可以像搭积木一样设计流程像调试普通程序一样调试AI行为最终交付真正能为用户创造价值的、稳定的AI应用。