LangGraph 状态机模型如何重构 AI Agent 的复杂逻辑

像经常用 LLM 的同学都知道现在最头疼的问题就是幻觉问题，在金融或精密计算领域，不确定性意味着风险。 如果 Agent 负责分析 NVDA 或 TSLA 的财报，开发者希望它在处理相同数据时，逻辑推导链条是严密的，而不是在不同时间给出自相矛盾的结论。或是需要 LLM 输出 JSON 来触发一个 API，我们不会希望 LLM 在 JSON 里多加了一个逗号或改变了字段名。

最后我还尝试用 LangGraph 的理念自己写了一个 smallest-LangGraph ，

LangGraph 可以做什么？

传统的 LangChain 核心逻辑是DAG（有向无环图）。我们可以轻松定义A -> B -> C的步骤，但如果你想让 AI 在B步骤发现结果不满意，自动跳回A重新执行，LangChain 的普通 Chain 很难优雅地实现。并且在复杂的长对话或多步骤任务中，维护一个全局的、可持久化的“状态快照”非常困难。

所以为了解决这些问题，LangGraph 就诞生了。LangGraph 的主要有这些核心优势：

1. 支持“循环（Cycles）”与“迭代”

   思考-> 2.行动-> 3.观察结果-> 4.如果不满意，回到第1步。 LangGraph 允许你定义这种闭环逻辑，这在长任务、自我修正代码、多轮调研场景下是刚需。

2. 状态管理

   LangGraph 引入了State的概念，所有节点共享同一个TypedDict，你可以精确定义哪些数据是追加的（operator.add），哪些是覆盖的。并且它可以自动保存每一步的状态。即使程序崩溃或需要人工审核，你也可以从特定的“存档点”恢复，而不需要从头运行。

3. 人机协作

   LangGraph 允许你将流程设计为“在某处强制停下”，等待人类信号后再继续。这在 LangChain 的线性模型中极难实现，但在 LangGraph 的状态机模型中只是一个节点属性。

4. 高度可控

   “如果工具返回报错，必须走 A 路径。” 这种确定性对于生产环境的后端服务至关重要。不能让模型乱输出，在生产环境上严格把控输出结果是很重要的。

LangGraph 结构

由于 LangGraph 的核心思想是将 Agent 的工作流建模为一张有向图（Directed Graph）。所以 LangGraph 有如下几个结构组成

• 全局状态（State）

  这个状态通常被定义为一个 Python 的TypedDict，它可以包含任何你需要追踪的信息，如对话历史、中间结果、迭代次数等，所有的节点都能读取和更新这个中心状态。

• 节点（Nodes）

  每个节点都是一个接收当前状态作为输入、并返回一个更新后的状态作为输出的 Python 函数。

• 边（Edges）

  边负责连接节点，定义工作流的方向。最简单的边是常规边，它指定了一个节点的输出总是流向另一个固定的节点。而 LangGraph 最强大的功能在于条件边（Conditional Edges）。它通过一个函数来判断当前的状态，然后动态地决定下一步应该跳转到哪个节点。

基于上面的概念，我们来做一个例子，假设我们要开发一个 Agent：它先翻译一段话，然后自己检查是否有语法错误，如果有，就打回重新翻译；如果没有，就结束。

首先，我们先定义状态 (State)：

from typing import TypedDict, Listclass AgentState(TypedDict): # 原始文本 input_text: str # 翻译后的文本 translated_text: str # 反思反馈 feedback: str # 循环次数（防止死循环） iterations: int

定义节点逻辑 (Nodes)：

def translator_node(state: AgentState): print("--- 正在翻译 ---") # 这里通常会调用 LLM new_text = f"Translated: {state['input_text']}" return {"translated_text": new_text, "iterations": state.get("iterations", 0) + 1}def critic_node(state: AgentState): print("--- 正在自检 ---") # 模拟检查逻辑，如果包含 'bad' 字符就认为不合格 if "bad" in state['translated_text']: return {"feedback": "发现不当词汇，请重试"} return {"feedback": "OK"}

定义路由逻辑 (Conditional Edges)：

def should_continue(state: AgentState): if state["feedback"] == "OK" or state["iterations"] > 3: return "end" else: return "rephrase"

构建图 (Graph Construction)：

from langgraph.graph import StateGraph, END# 1. 初始化图workflow = StateGraph(AgentState)# 2. 添加节点workflow.add_node("translator", translator_node)workflow.add_node("critic", critic_node)# 3. 设置入口点workflow.set_entry_point("translator")# 4. 连接节点workflow.add_edge("translator", "critic")# 5. 添加条件边 (根据 critic 的反馈决定去向)workflow.add_conditional_edges( "critic", should_continue, { "rephrase": "translator", # 如果不 OK，回到翻译节点 "end": END # 如果 OK，结束 })# 6. 编译成可执行应用app = workflow.compile()

通过上面这种编排方式，可以让 LLM 概率性输出产生确定性的输出，通过各种限制节点，很好的控制了 LLM 的访问的节点。

下面我给出完整的例子，大家可以用这个例子去尝试一下：

from typing import TypedDict, Listfrom langchain_openai import ChatOpenAIfrom langgraph.graph import StateGraph, ENDfrom langchain_core.messages import SystemMessage, HumanMessagellm = ChatOpenAI( temperature=0.6, model="glm-4.6V", openai_api_key="", openai_api_base="https://open.bigmodel.cn/api/paas/v4/")class AgentState(TypedDict): # 原始文本 input_text: str # 翻译后的文本 translated_text: str # 反思反馈 feedback: str # 循环次数（防止死循环） iterations: intdef translator_node(state: AgentState): """翻译节点：负责将中文翻译成英文""" print(f"\n--- [节点：翻译器] 第 {state.get('iterations', 0) + 1} 次尝试 ---") iters = state.get("iterations", 0) feedback = state.get("feedback", "无") # 构建提示词：如果是重试，带上反馈建议 system_prompt = "你是一个专业的翻译官。请将用户的中文翻译成地道、优雅的英文。" if iters > 0: system_prompt += f" 注意：这是第二次尝试，请参考之前的反馈进行改进：{feedback}" response = llm.invoke([ SystemMessage(content=system_prompt), HumanMessage(content=state["input_text"]) ]) return { "translated_text": response.content, "iterations": iters + 1 }def critic_node(state: AgentState): """评审节点：检查翻译质量""" print("--- [节点：评审员] 正在检查翻译质量... ---") system_prompt = ( "你是一个严苛的英文编辑。请评价以下翻译是否准确、地道。" "如果翻译得很好，请只回复关键词：【PASS】。" "如果翻译有改进空间，请直接指出问题并给出改进建议。" ) user_content = f"原文：{state['input_text']}\n译文：{state['translated_text']}" response = llm.invoke([ SystemMessage(content=system_prompt), HumanMessage(content=user_content) ]) return {"feedback": response.content}# 4. 定义路由逻辑def should_continue(state: AgentState): """判断是继续修改还是直接结束""" if"【PASS】"in state["feedback"] or state["iterations"] >= 3: if state["iterations"] >= 3: print("!!! 达到最大尝试次数，停止优化。") return"end" else: print(f">>> 反馈建议：{state['feedback']}") return"rephrase"# 1. 初始化图workflow = StateGraph(AgentState)# 2. 添加节点workflow.add_node("translator", translator_node)workflow.add_node("critic", critic_node)# 3. 设置入口点workflow.set_entry_point("translator")# 4. 连接节点workflow.add_edge("translator", "critic")# 5. 添加条件边 (根据 critic 的反馈决定去向)workflow.add_conditional_edges( "critic", should_continue, { "rephrase": "translator", # 如果不 OK，回到翻译节点 "end": END # 如果 OK，结束 })# 6. 编译成可执行应用app = workflow.compile()# 7. 运行时交互if __name__ == "__main__": print("=== LangGraph 智能翻译 Agent (输入 'exit' 退出) ===") whileTrue: user_input = input("\n请输入想要翻译的中文内容: ") if user_input.lower() == 'exit': break # 初始状态 initial_state = { "input_text": user_input, "iterations": 0 } # 运行图并获取最终状态 final_state = app.invoke(initial_state) print("\n" + "=" * 30) print(f"最终翻译结果：\n{final_state['translated_text']}") print("=" * 30)

LangGraph 是如何管理状态的？

State Reducer 自动合并 state

Reducer 在 LangGraph 中就是一种更新状态的处理逻辑，如果没有指定默认行为是用新值覆盖旧值。想要指定 Reducer 只需要通过 typing.Annotated 字段绑定一个 Reducer 函数即可。

比如使用 operator.add 定义这是一个“追加型”字段：

from typing import Annotated, TypedDictfrom langgraph.graph import StateGraph, ENDimport operator# 定义状态结构 (类似 Go 的 Struct)class AgentState(TypedDict): # 使用 Annotated 和 operator.add 定义这是一个“追加型”字段 # 每次节点返回消息，都会 append 到这个列表，而不是覆盖它 messages: Annotated[list[str], operator.add] # 普通字段，默认行为是 Overwrite (覆盖) # 适合存储状态机当前的步骤或分析结论 current_status: str # 计数器，也可以使用 operator.add 实现增量累加 retry_count: Annotated[int, operator.add]

Checkpointer + Thread 持久化状态

在 LangGraph 中，Checkpointer是一个持久化层接口，这意味着历史的对话记录，可以被自动持久化到数据库（如 SQLite 或其他外部数据库）中。这使得即使应用程序重启或用户断开连接，对话历史也能被保存和恢复，从而实现“真正的多轮记忆”。

LangGraph 提供了多种 Checkpointer 以便应对不同的使用场景：

• MemorySaver 保存在内存，适用开发调试、单元测试；
• SqliteSaver 保存在本地的.db文件，轻量级应用、边缘计算适合单机部署；
• PostgresSaver 保存在 PostgreSQL，适合用在生产环境、多实例部署；
• RedisSaver 适合处理高频、短时会话；

LangGraph 通过 thread_id 会话的唯一标识，结合 Checkpointer 就可以实现状态的隔离：

首先指定一个 指定一个thread_id，所有相关的状态都会被保存到这个线程中。

config = {"configurable": {"thread_id": "conversation_1"}}graph.invoke(input_data, config)

编译的时候传入 Checkpointer 即可。

# 创建 checkpointercheckpointer = InMemorySaver()# 编译图时传入 checkpointergraph = builder.compile(checkpointer=checkpointer)

完整示例：

from langgraph.checkpoint.memory import InMemorySaverfrom langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END, MessagesState llm = ChatOpenAI( temperature=0.6, model="glm-4.6v", openai_api_key="", openai_api_base="https://open.bigmodel.cn/api/paas/v4/")# 定义节点函数def call_model(state: MessagesState): response = llm.invoke(state["messages"]) return {"messages": response}# 构建图builder = StateGraph(MessagesState)builder.add_node("agent", call_model)builder.add_edge(START, "agent")# 创建 checkpointercheckpointer = InMemorySaver()# 编译图时传入 checkpointergraph = builder.compile(checkpointer=checkpointer)# 第一次对话config = {"configurable": {"thread_id": "user_123"}}response1 = graph.invoke( {"messages": [{"role": "user", "content": "你好，我的名字是张三"}]}, config)print(f"AI: {response1['messages'][-1].content}")# 第二次对话（相同 thread_id）response2 = graph.invoke( {"messages": [{"role": "user", "content": "我的名字是什么？"}]}, config # 使用相同的 thread_id)print(f"AI: {response2['messages'][-1].content}")# 获取当前的状态信息print(f"AI: {graph.get_state(config)}")

除此之外，可以graph.get_state()/graph.get_state_history()拿到当前/历史状态；也可以基于 checkpoint 做 replay、update_state（时间旅行能力通常要求启用 checkpointer）。

Super-step 原子循环单元

由于一个 node 也可以连接多个 node，多个 node 也可以连接到 一个 node，所以 LangGraph 设计了 Super-step 来作为原子循环单元。比如下面的例子：

graph.set_entry_point("n1") graph.add_edge("n1", "n2") graph.add_edge("n1", "n3") graph.add_edge("n2", "n4") graph.add_edge("n3", "n4") graph.add_edge("n4", END)

LangGraph 只分了三步就执行完了该循环。如下图，第二步的时候会 n2、n3 节点并行执行。

graph active nodes in each superstep

并且每个 super-step 都会自动保存一个 checkpoint，这就是持久化机制的基础。即使程序中断，也能从最后一个 super-step 的 checkpoint 恢复执行。

Human-in-the-loop 人机协同

Human-in-the-loop 本质上就是让 agent “关键时刻”暂停，它的底层靠的是 **interrupt + 持久化(checkpoint)**：暂停时把状态存起来，恢复时从存档续跑。

比如我们想要是线一个场景就是让 AI 去判断是否应该要人工审核，如过需要人工审核，那么就 interrupt 进行中断，然后等人工输入之后根据执行逻辑进行恢复，然后配合Command(resume=...)恢复。

基本流程可以是这样：

import uuidfrom langgraph.types import interrupt, Commanddef ask_human(state): answer = interrupt("Do you approve?") return {"approved": answer}config = {"configurable": {"thread_id": str(uuid.uuid4())}}# 第一次跑：会中断，返回 __interrupt__graph.invoke({"input": "x"}, config=config)# 人给了答复后：用 Command(resume=...) 恢复graph.invoke(Command(resume=True), config=config)

这个例子中interrupt()会暂停图执行，把一个值（必须可 JSON 序列化）抛给调用方，并依赖 checkpointer 持久化状态；然后你用同一个thread_id重新调用图，并传入Command(resume=...)来继续。

接下来我们看一个完整的例子，设计一个常见的场景，当模型觉得需要“找专家/找人类”时，会调用一个工具human_assistance，而这个工具会用interrupt()把流程暂停下来，等你在命令行里输入专家建议后，再用Command(resume=...)把图唤醒继续跑。

from langgraph.checkpoint.memory import InMemorySaverfrom langchain_openai import ChatOpenAIfrom typing_extensions import TypedDictfrom typing import Annotatedfrom langgraph.graph import StateGraph, START, ENDfrom langgraph.graph.message import add_messagesfrom langgraph.types import Command, interruptfrom langchain_core.tools import toolfrom langgraph.prebuilt import ToolNode, tools_condition from langchain_core.messages import HumanMessagellm = ChatOpenAI( temperature=0.6, model="glm-4.6v", openai_api_key="", openai_api_base="https://open.bigmodel.cn/api/paas/v4/")class State(TypedDict): messages: Annotated[list, add_messages]@tooldef human_assistance(query: str) -> str: """Request assistance from a human.""" human_response = interrupt({"query": query}) return human_response["data"]tools = [human_assistance]llm_with_tools = llm.bind_tools(tools)def chatbot(state: State): message = llm_with_tools.invoke(state["messages"]) return {"messages": [message]}tool_node = ToolNode(tools=tools)graph_builder = StateGraph(State)graph_builder.add_node("tools", tool_node)graph_builder.add_conditional_edges( "chatbot", tools_condition,)graph_builder.add_node("chatbot", chatbot)graph_builder.add_edge(START, "chatbot")graph_builder.add_edge("chatbot", END)memory = InMemorySaver()graph = graph_builder.compile(checkpointer=memory)config = {"configurable": {"thread_id": "test_thread_123"}}# 第一步：用户提出一个需要“人工协助”的问题print("--- 第一阶段：AI 运行并遇到 interrupt ---")initial_input = HumanMessage(content="你好，帮我找个专家回答我的问题")for event in graph.stream({"messages": [initial_input]}, config, stream_mode="values"): if"messages"in event: event["messages"][-1].pretty_print()# 此时，你会发现程序停止了，因为它卡在 `human_assistance` 的 `interrupt` 处。# 第二步：模拟人类（你）在一段时间后看到了请求并回复print("\n--- 第二阶段：模拟人类介入并提供答案 ---")# 我们构造一个 Command 对象来“唤醒”它# resume 里的内容会直接成为 interrupt() 函数的返回值expert_input = input("专家建议: ")human_feedback = {"data": expert_input}for event in graph.stream( Command(resume=human_feedback), # 这里是恢复运行的关键 config, stream_mode="values"): if"messages"in event: event["messages"][-1].pretty_print()snapshot = graph.get_state(config)print(snapshot.values)

LangGraph如何轻松实现 Agent 多种执行范式

ReAct

ReAct由Shunyu Yao于2022年提出[1]，其核心思想是模仿人类解决问题的方式，将推理 (Reasoning)与行动 (Acting)显式地结合起来，形成一个“思考-行动-观察”的循环。

img

ReAct范式通过一种特殊的提示工程来引导模型，使其每一步的输出都遵循一个固定的轨迹：

• Thought (思考)：这是智能体的“内心独白”。它会分析当前情况、分解任务、制定下一步计划，或者反思上一步的结果。
• Action (行动)：这是智能体决定采取的具体动作，通常是调用一个外部工具API 。
• Observation (观察)：这是执行Action后从外部工具返回的结果，例如或API的返回值。

智能体将不断重复这个Thought -> Action -> Observation的循环，将新的观察结果追加到历史记录中，形成一个不断增长的上下文，直到它在Thought中认为已经找到了最终答案，然后输出结果。

from typing import Annotated, Literalfrom typing_extensions import TypedDictfrom langgraph.graph import StateGraph, ENDfrom langgraph.graph.message import add_messagesfrom langgraph.prebuilt import ToolNodefrom langchain_openai import ChatOpenAIfrom langchain_core.tools import tool# --- 1. 双手：定义查天气的工具 ---@tooldef get_weather(city: str): """查询指定城市的天气""" # 这里模拟后端 API 返回数据 if"北京"in city: return"晴天，25度" return"阴天，20度"tools = [get_weather]tool_node = ToolNode(tools)# --- 2. 记忆：定义存储对话的状态 ---class State(TypedDict): messages: Annotated[list, add_messages]# --- 3. 大脑：定义思考逻辑 ---model = ChatOpenAI( temperature=0.6, model="glm-4.6v", openai_api_key="", openai_api_base="https://open.bigmodel.cn/api/paas/v4/").bind_tools(tools)def call_model(state: State): # 大脑看一眼目前的对话，决定是直接说话还是去用手拿工具 return {"messages": [model.invoke(state["messages"])]}# --- 4. 路由：判断下一步是干活还是结束 ---def should_continue(state: State): last_message = state["messages"][-1] # 如果大脑发出的指令包含“调用工具”，就去 tools 节点 if last_message.tool_calls: return"tools" # 如果大脑直接说话了，就结束 return END# --- 5. 编排图（把脑和手连起来） ---workflow = StateGraph(State)workflow.add_node("agent", call_model)workflow.add_node("tools", tool_node)workflow.set_entry_point("agent")# 条件边：agent 运行完，判断是去 tools 还是结束workflow.add_conditional_edges("agent", should_continue)# 普通边：tools 运行完（干完活了），必须把结果拿回给 agent 看workflow.add_edge("tools", "agent")app = workflow.compile()# --- 6. 执行测试 ---for chunk in app.stream({"messages": [("user", "北京今天天气怎么样？")]}): print(chunk)

Plan-and-Solve

Plan-and-Solve 顾名思义，这种范式将任务处理明确地分为两个阶段：**先规划 (Plan)，后执行 (Solve)**。Plan-and-Solve Prompting 由 Lei Wang 在2023年提出。其核心动机是为了解决思维链在处理多步骤、复杂问题时容易“偏离轨道”的问题。

Plan-and-Solve 将整个流程解耦为两个核心阶段：

1. 规划阶段 (Planning Phase)： 首先，智能体会接收用户的完整问题。它的第一个任务不是直接去解决问题或调用工具，而是将问题分解，并制定出一个清晰、分步骤的行动计划。
2. 执行阶段 (Solving Phase)： 在获得完整的计划后，智能体进入执行阶段。它会严格按照计划中的步骤，逐一执行。每一步的执行都可能是一次独立的 LLM 调用，或者是对上一步结果的加工处理，直到计划中的所有步骤都完成，最终得出答案。

img

import operatorfrom typing import Annotated, List, Tuple, Unionfrom typing_extensions import TypedDictfrom pydantic import BaseModel, Fieldfrom langgraph.graph import StateGraph, START, ENDfrom langchain_openai import ChatOpenAI# 1. 定义状态 (State)class PlanExecuteState(TypedDict): input: str # 原始问题 plan: List[str] # 当前待办清单 past_steps: Annotated[List[Tuple], operator.add] # 已完成的步骤和结果 response: str # 最终答案# 2. 定义结构化输出模型 (用于 Planner)class Plan(BaseModel): """步骤清单""" steps: List[str] = Field(description="为了回答问题需要执行的步骤")# 3. 定义节点逻辑model = ChatOpenAI( temperature=0.6, model="glm-4.6v", openai_api_key="", openai_api_base="https://open.bigmodel.cn/api/paas/v4/")planner_model = model.with_structured_output(Plan, method="function_calling")# --- 节点 A: 规划者 ---def planner_node(state: PlanExecuteState): plan = planner_model.invoke(f"针对以下问题制定计划: {state['input']}") return {"plan": plan.steps}# --- 节点 B: 执行者 (这里简化了工具调用) ---def executor_node(state: PlanExecuteState): step = state["plan"][0] # 取当前第一步 print(f"--- 正在执行: {step} ---") # 模拟工具执行结果 result = f"已完成 {step} 的查询，结果为: [模拟数据]" return {"past_steps": [(step, result)], "plan": state["plan"][1:]}# --- 节点 C: 重规划者 (决定是继续还是结束) ---def replanner_node(state: PlanExecuteState): ifnot state["plan"]: # 如果清单空了，让 AI 生成最终总结 summary = model.invoke( f"请基于已完成的步骤和结果给出最终答案：{state['past_steps']}" ) return {"response": summary.content} return {"response": None}# 4. 路由逻辑def should_continue(state: PlanExecuteState): if state["response"]: return END return"executor"# 5. 编排图workflow = StateGraph(PlanExecuteState)workflow.add_node("planner", planner_node)workflow.add_node("executor", executor_node)workflow.add_node("re-planner", replanner_node)workflow.set_entry_point("planner")workflow.add_edge("planner", "executor")workflow.add_edge("executor", "re-planner")# 循环逻辑：根据 re-planner 的判断决定是否回 executorworkflow.add_conditional_edges("re-planner", should_continue)app = workflow.compile()# 6. 测试input_query = {"input": "对比北京和上海的天气，哪个更热？"}for event in app.stream(input_query): print(event)

Reflection

Reflection 机制的核心思想，正是为智能体引入一种事后（post-hoc）的自我校正循环，使其能够像人类一样，审视自己的工作，发现不足，并进行迭代优化。 Reflection 框架是Shinn, Noah 在2023年提出，其核心工作流程可以概括为一个简洁的三步循环：执行 -> 反思 -> 优化。

1. **执行 (Execution)**：首先，智能体使用我们熟悉的方法（如 ReAct 或 Plan-and-Solve）尝试完成任务，生成一个初步的解决方案或行动轨迹。这可以看作是“初稿”。
2. **反思 (Reflection)**：接着，智能体进入反思阶段。它会调用一个独立的、或者带有特殊提示词的大语言模型实例，来扮演一个“评审员”的角色。这个“评审员”会审视第一步生成的“初稿”，并从多个维度进行评估，例如：

• 事实性错误：是否存在与常识或已知事实相悖的内容？
• 逻辑漏洞：推理过程是否存在不连贯或矛盾之处？
• 效率问题：是否有更直接、更简洁的路径来完成任务？
• 遗漏信息：是否忽略了问题的某些关键约束或方面？ 根据评估，它会生成一段结构化的**反馈 (Feedback)**，指出具体的问题所在和改进建议。

3. **优化 (Refinement)**：最后，智能体将“初稿”和“反馈”作为新的上下文，再次调用大语言模型，要求它根据反馈内容对初稿进行修正，生成一个更完善的“修订稿”。

img

from typing import TypedDictfrom langchain_openai import ChatOpenAIfrom langgraph.graph import StateGraph, START, ENDclass ReflectionState(TypedDict): prompt: str draft: str critique: str final: str iteration: intllm = ChatOpenAI( temperature=0.6, model="glm-4.6v", openai_api_key="", openai_api_base="https://open.bigmodel.cn/api/paas/v4/",)MAX_ITERS = 2def generate_draft(state: ReflectionState): msg = llm.invoke(f"请写一段简短答案：{state['prompt']}") return {"draft": msg.content, "iteration": 0}def reflect_on_draft(state: ReflectionState): prompt = ( "你是严格的审稿人。请指出这段答案的问题并给出改进建议。" "如果没有明显问题，请只输出 NO_ISSUES。\n\n" f"答案:\n{state['draft']}" ) critique = llm.invoke(prompt) print(f"--- 正在执行: reflect，critique:\n {critique.content} ---") return {"critique": critique.content}def revise_draft(state: ReflectionState): prompt = ( "请根据以下反馈重写答案，保持简短清晰：\n\n" f"反馈:\n{state['critique']}\n\n" f"原答案:\n{state['draft']}" ) revision = llm.invoke(prompt) print(f"--- 正在执行: revise，原答案:\n{state['draft']}，改进后：\n{revision.content} ---") return {"draft": revision.content, "iteration": state["iteration"] + 1}def finalize(state: ReflectionState): return {"final": state["draft"]}def should_reflect(state: ReflectionState): if state["critique"].strip() == "NO_ISSUES": return"finalize" if state["iteration"] >= MAX_ITERS: return"finalize" return"revise"workflow = StateGraph(ReflectionState)workflow.add_node("generate", generate_draft)workflow.add_node("reflect", reflect_on_draft)workflow.add_node("revise", revise_draft)workflow.add_node("finalize", finalize)workflow.add_edge(START, "generate")workflow.add_edge("generate", "reflect")workflow.add_conditional_edges("reflect", should_reflect)workflow.add_edge("revise", "reflect")workflow.add_edge("finalize", END)app = workflow.compile()if __name__ == "__main__": input_data = {"prompt": "用三句话解释什么是 LangGraph。"} result = app.invoke(input_data) print("最终答案:") print(result["final"])

大致流程就是，首先里面需要有两个角色：写稿人和审稿人，然后用 should_reflect 来判断是否需要重写，然后用 MAX_ITERS 来限制一下最大撰写次数。

[START] | v (generate_draft) | v (reflect_on_draft) -- NO_ISSUES --> (finalize) --> [END] | | iteration >= MAX_ITERS +-----------------------> (finalize) --> [END] | +-- else --> (revise_draft) --+ | v (reflect_on_draft)

Multi-Agent Pattern

Multi-Agent 模式是将复杂的任务拆解为多个专门化、独立且可协同的微服务，每个服务（Agent）只负责一个特定的领域。

因为单个 Prompt 包含太多工具和指令会导致 LLM “迷失”，模型表现下降。所以通过使用doge Agent 进行职责分离，不同的 Agent 可以使用不同的 Prompt、不同的模型（如 GPT-4o 负责决策，Llama-3 负责写代码），甚至不同的工具集。

比如下面的例子中：一个“PM” Agent 负责拆解任务，并将子任务分发给不同的“员工（Workers）”。

img

img

from typing import Annotated, Literalfrom typing_extensions import TypedDictfrom langchain_openai import ChatOpenAIfrom langgraph.graph import StateGraph, START, ENDfrom langgraph.graph.message import add_messages# Multi-agent pattern: a supervisor routes work between specialist agents.llm = ChatOpenAI( temperature=0.4, model="glm-4.6v", openai_api_key="", openai_api_base="https://open.bigmodel.cn/api/paas/v4/",)class State(TypedDict): messages: Annotated[list, add_messages] next: str turn: intMAX_TURNS = 6def _call_agent(system_prompt: str, messages: list, name: str): response = llm.invoke([{"role": "system", "content": system_prompt}] + messages) return { "messages": [ {"role": "assistant", "name": name, "content": response.content} ], "turn": 1, }def supervisor(state: State): if state["turn"] >= MAX_TURNS: return {"next": "finish"} system = ( "You are a supervisor managing a team: researcher, writer, critic. " "Choose who should act next or finish. " "Respond with exactly one word: researcher, writer, critic, finish." ) response = llm.invoke([{"role": "system", "content": system}] + state["messages"]) decision = response.content.strip().lower() for option in ("researcher", "writer", "critic", "finish"): if option in decision: return {"next": option} return {"next": "finish"}def researcher(state: State): system = ( "You are a researcher. Gather key facts and constraints for the task. " "Be concise and list only essential points." ) return _call_agent(system, state["messages"], "researcher")def writer(state: State): system = ( "You are a writer. Produce a clear, structured response using the context. " "If facts are missing, note assumptions." ) return _call_agent(system, state["messages"], "writer")def critic(state: State): system = ( "You are a critic. Identify gaps, risks, or unclear parts in the draft, " "then suggest improvements." ) return _call_agent(system, state["messages"], "critic")def route_next(state: State) -> Literal["researcher", "writer", "critic", "finish"]: return state["next"]builder = StateGraph(State)builder.add_node("supervisor", supervisor)builder.add_node("researcher", researcher)builder.add_node("writer", writer)builder.add_node("critic", critic)builder.add_edge(START, "supervisor")builder.add_conditional_edges( "supervisor", route_next, { "researcher": "researcher", "writer": "writer", "critic": "critic", "finish": END, },)builder.add_edge("researcher", "supervisor")builder.add_edge("writer", "supervisor")builder.add_edge("critic", "supervisor")app = builder.compile()if __name__ == "__main__": user_input = "创建一款广告招商的帖子" initial_state = { "messages": [{"role": "user", "content": user_input}], "next": "supervisor", "turn": 0, } for event in app.stream(initial_state): for value in event.values(): if"messages"in value: msg = value["messages"][-1] name = msg.get("name", "assistant") print(f"[{name}] {msg['content']}")

想入门 AI 大模型却找不到清晰方向？备考大厂 AI 岗还在四处搜集零散资料？别再浪费时间啦！2025 年AI 大模型全套学习资料已整理完毕，从学习路线到面试真题，从工具教程到行业报告，一站式覆盖你的所有需求，现在全部免费分享！

👇👇扫码免费领取全部内容👇👇

​

一、学习必备：100+本大模型电子书+26 份行业报告 + 600+ 套技术PPT，帮你看透 AI 趋势

想了解大模型的行业动态、商业落地案例？大模型电子书？这份资料帮你站在 “行业高度” 学 AI：

1. 100+本大模型方向电子书

2. 26 份行业研究报告：覆盖多领域实践与趋势

报告包含阿里、DeepSeek 等权威机构发布的核心内容，涵盖：

• 职业趋势：《AI + 职业趋势报告》《中国 AI 人才粮仓模型解析》；
• 商业落地：《生成式 AI 商业落地白皮书》《AI Agent 应用落地技术白皮书》；
• 领域细分：《AGI 在金融领域的应用报告》《AI GC 实践案例集》；
• 行业监测：《2024 年中国大模型季度监测报告》《2025 年中国技术市场发展趋势》。

3. 600+套技术大会 PPT：听行业大咖讲实战

PPT 整理自 2024-2025 年热门技术大会，包含百度、腾讯、字节等企业的一线实践：

• 安全方向：《端侧大模型的安全建设》《大模型驱动安全升级（腾讯代码安全实践）》；
• 产品与创新：《大模型产品如何创新与创收》《AI 时代的新范式：构建 AI 产品》；
• 多模态与 Agent：《Step-Video 开源模型（视频生成进展）》《Agentic RAG 的现在与未来》；
• 工程落地：《从原型到生产：AgentOps 加速字节 AI 应用落地》《智能代码助手 CodeFuse 的架构设计》。

二、求职必看：大厂 AI 岗面试 “弹药库”，300 + 真题 + 107 道面经直接抱走

想冲字节、腾讯、阿里、蔚来等大厂 AI 岗？这份面试资料帮你提前 “押题”，拒绝临场慌！

1. 107 道大厂面经：覆盖 Prompt、RAG、大模型应用工程师等热门岗位

面经整理自 2021-2025 年真实面试场景，包含 TPlink、字节、腾讯、蔚来、虾皮、中兴、科大讯飞、京东等企业的高频考题，每道题都附带思路解析：

2. 102 道 AI 大模型真题：直击大模型核心考点

针对大模型专属考题，从概念到实践全面覆盖，帮你理清底层逻辑：

3. 97 道 LLMs 真题：聚焦大型语言模型高频问题

专门拆解 LLMs 的核心痛点与解决方案，比如让很多人头疼的 “复读机问题”：

三、路线必明： AI 大模型学习路线图，1 张图理清核心内容

刚接触 AI 大模型，不知道该从哪学起？这份「AI大模型 学习路线图」直接帮你划重点，不用再盲目摸索！

路线图涵盖 5 大核心板块，从基础到进阶层层递进：一步步带你从入门到进阶，从理论到实战。

L1阶段:启航篇丨极速破界AI新时代

L1阶段：了解大模型的基础知识，以及大模型在各个行业的应用和分析，学习理解大模型的核心原理、关键技术以及大模型应用场景。

L2阶段：攻坚篇丨RAG开发实战工坊

L2阶段：AI大模型RAG应用开发工程，主要学习RAG检索增强生成：包括Naive RAG、Advanced-RAG以及RAG性能评估，还有GraphRAG在内的多个RAG热门项目的分析。

L3阶段：跃迁篇丨Agent智能体架构设计

L3阶段：大模型Agent应用架构进阶实现，主要学习LangChain、 LIamaIndex框架，也会学习到AutoGPT、 MetaGPT等多Agent系统，打造Agent智能体。

L4阶段：精进篇丨模型微调与私有化部署

L4阶段：大模型的微调和私有化部署，更加深入的探讨Transformer架构，学习大模型的微调技术，利用DeepSpeed、Lamam Factory等工具快速进行模型微调，并通过Ollama、vLLM等推理部署框架，实现模型的快速部署。

L5阶段：专题集丨特训篇 【录播课】

四、资料领取：全套内容免费抱走，学 AI 不用再找第二份

不管你是 0 基础想入门 AI 大模型，还是有基础想冲刺大厂、了解行业趋势，这份资料都能满足你！
现在只需按照提示操作，就能免费领取：

👇👇扫码免费领取全部内容👇👇

​

2025 年想抓住 AI 大模型的风口？别犹豫，这份免费资料就是你的 “起跑线”！