LLM Agent 设计模式：从 ReAct 到多智能体协作

代码关键细节

CodeAct是 ReAct 的演进形态，不再预定义工具函数，而是让 LLM 根据任务动态生成 Python 代码，由沙箱环境执行并返回结果。

SYSTEM_PROMPT = """你是一个能够编写和执行代码的智能助手。
当用户提出问题时：
1. 分析问题并确定需要编写什么代码
2. 编写能解决问题的 Python 代码
3. 使用 execute_python 工具执行代码
4. 分析执行结果，如有错误则修改代码再次执行
5. 最终给用户提供答案"""

@tool
def execute_python(code: str) -> str:
    """执行 Python 代码并返回结果"""
    try:
        local_vars = {}
        exec(code, {}, local_vars)
        return str(local_vars.get('result', '执行成功'))
    except Exception as e:
        return f"Error: {str(e)}"

class OverallState(TypedDict):
    messages: Annotated[list, add_messages]
    actions: list[dict]
    output: str
    user_prompt: str

# LangGraph 状态图定义
class CodeActGraph():
    def run(self, user_prompt: str):
        def _create_agent_action(ai_message):
            # 提取代码部分
            content = ai_message.content
            if "```python" in content:
                code_blocks = content.split("```python")
                code = code_blocks[1].split("```")[0].strip()
                return {"tool": "execute_python", "tool_input": code}
            return None

        def llm_call(state: OverallState):
            # 构建包含系统提示词和用户提示词的消息列表
            messages = [
                SystemMessage(content=SYSTEM_PROMPT),
                HumanMessage(content=state["user_prompt"])
            ]+state["messages"]

            ret = llm_Tongyi.invoke(messages)

            action = _create_agent_action(ret)
            if action:
                return {"messages": [ret], "actions": [action], "output": ""}
            return {"messages": [ret], "output": ret.content}

        def enter_process(state: OverallState):
            if state["output"] != "":
                return END
            return "process_node"

        def process_node(state: OverallState):
            actions = state["actions"]
            for action in actions:
                tool_name = action["tool"]
                tool_input = action["tool_input"]
                tool_fn = next(t for t in tools if t.name == tool_name)
                observation = tool_fn.invoke(tool_input)  #执行工具
                state["messages"].append(HumanMessage(content=f"##执行结果:\n{observation}"))

            return state

        graph = StateGraph(OverallState)
        graph.add_node("llm_call", llm_call)
        graph.add_node("enter_process", enter_process)
        graph.add_node("process_node", process_node)

        graph.add_edge(START, "llm_call")
        graph.add_conditional_edges("llm_call",enter_process)
        graph.add_edge("process_node", "llm_call")

        agent = graph.compile()
        ret = agent.invoke(OverallState(user_prompt=user_prompt))

        return ret["output"]

START → LLM 分析/生成代码 → 执行代码 → 观察结果 → (循环) → 最终答案
         ↑                                          |
         └──────────────────────────────────────────┘

计划模式将任务分为生成计划和执行计划两个阶段，根据计划更新策略分为两个版本：

• 核心思想：LLM 一次性生成完整计划，后续严格按计划步骤执行。

• 计划生成： LLM根据用户问题一次性生成完整计划（字符串形式）。

• 执行： 模型根据计划逐步执行，可能调用工具，直到输出包含“Final Answer”为止。

• 进度跟踪： 通过消息历史记录已执行的步骤和工具结果，模型自主判断下一步。

PLAN_PROMPT = """
你是一个任务规划助手。请为用户提出的问题创建分析方案步骤：
1. 用中文列出清晰步骤
2. 每个步骤标记序号
3. 明确说明需要分析和执行的内容
4. 设计的方案步骤要紧紧贴合工具所能返回的内容
"""

class PlanState(MessagesState):
    plan: str

def plan_node(state: PlanState):
    # 调用 LLM
    response = llm_Tongyi.invoke([SystemMessage(content=PLAN_PROMPT),state["messages"][0]])

    state["plan"] = response.content
    print("计划：\n" + state["plan"])
    return state

def execute_node(state: PlanState):
    messages = [
        SystemMessage(
            content=PLAN_EXECUTE_PROMPT.format(plan=state["plan"])
        )
    ] + state["messages"]

    response = llm_with_tools.invoke(messages)
    print("执行：\n" + response.content)
    state["messages"].append(response)

    return state

def tool_node(state: PlanState):
    for tool_call in state["messages"][-1].tool_calls:
        print(tool_call)
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        state["messages"].append(ToolMessage(content=observation, tool_call_id=tool_call["id"]))
    return state

def should_continue(state: PlanState) -> Literal["tool_node", "END"]:
    messages = state["messages"]
    last_message = messages[-1]
    if "Final Answer" in last_message.content:
        return "END"
    return "tool_node"

agent_builder = StateGraph(PlanState)

# Add nodes
agent_builder.add_node("plan_node", plan_node)
agent_builder.add_node("execute_node", execute_node)
agent_builder.add_node("tool_node", tool_node)

# Add edges
agent_builder.add_edge(START, "plan_node")
agent_builder.add_edge("plan_node", "execute_node")
agent_builder.add_conditional_edges(
    "execute_node",
    should_continue,
    {
        "tool_node": "tool_node",
        "END": END,
    },
)
agent_builder.add_edge("tool_node", "execute_node")

agent = agent_builder.compile()

ret = agent.invoke({"plan":"", "messages": [HumanMessage(content="茅台和青岛啤酒哪个贵？")]})
print(ret["messages"][-1].content)

print(ret["messages"])

大模型通过消息历史（state["messages"]）获取执行进度。

1. 状态管理：PlanState 继承自 MessagesState，维护完整的消息历史。

2. 消息历史传递：在 execute_node 中，每次调用 LLM 都会传入完整的消息历史。

3. 消息累积过程：

• 初始：[HumanMessage("茅台和青岛啤酒哪个贵？")]

• 第一次 execute_node：LLM 看到用户问题，生成工具调用（AIMessage with tool_calls）。

• tool_node：执行工具，将结果追加为 ToolMessage。

• 第二次 execute_node：LLM 看到用户问题、之前的工具调用和工具结果，继续执行。

4. 工具结果追加：tool_node 将工具执行结果追加到消息历史。

5. 执行流程循环：

plan_node → execute_node → tool_node → execute_node → tool_node → ... → END

每次循环，LLM 都能看到：

• 用户原始问题

• 之前所有的工具调用（AIMessage with tool_calls）

• 所有工具的执行结果（ToolMessage）

• 之前的执行步骤和中间结果

因此，LLM 通过完整的消息历史知道：

• 已执行哪些步骤

• 工具返回的结果

• 当前处于计划的哪一步

• 下一步该做什么

• 计划生成： 模型在每步执行后动态更新计划，移除已完成步骤，只保留剩余步骤。

• 执行： 每次只执行计划中的第一步（plan[0]），粒度更细，控制更精确。

• 进度跟踪： 使用结构化状态（past_steps 记录已完成步骤及其结果），并在计划更新时由模型判断是否直接输出答案。

prompt：

SYSTEM_PROMPT = """
你是一个任务执行助手
"""

PLAN_PROMPT = """ 
你是一个任务执行助手,根据给定的任务按步骤执行,不要添加任何多余的步骤 
最后一步的结果应该是最终答案。确保每个步骤都有所需的所有信息，不要跳过步骤

你的目标或问题:
{input}

你的计划是:
{plan}

当前已经完成的步骤:
{past_steps}

#请遵循以下规则
1.请输出JSON字符串,
2.相应地更新你的计划。如果不需要再执行更多步骤,就直接回复用户。否则，请填写计划。
3.只在计划中添加仍然需要完成的步骤。不要将之前完成的步骤作为计划的一部分返回。

#输出格式说明：
- 如果所有步骤都完成了，输出：{{"actions": {{"response": "最终回答内容"}}}}
- 如果还有步骤需要执行，输出：{{"actions": {{"steps": ["步骤1", "步骤2", ...]}}}}

注意：actions 字段是必需的，不要使用其他字段名。
"""

提示词，设计中“最后一步的结果应该是最终答案。确保每个步骤都有所需的所有信息，不要跳过步骤”，避免了之前的使用Final Answer。

执行完的步骤会放在{past_steps}中，未执行的后续步骤，放在{{"actions": {{"steps": ["步骤1", "步骤2", ...]}}}}中为什么是这么设计格式，因为要与Action匹配。

class Action(BaseModel):
    """执行的动作"""
    actions: Union[Response, Plan] = Field(
        description="要执行的动作,如果直接进行输出则选择Response,"
                    "如果需执行工具,请选择Plan"
    )

核心流程：

class PlanAgent():
    def __init__(self,prompt_sys,plan=[],tools=[]):
        self.tools = tools
        llm=DeepSeek()
        llm2=Tongyi()
        self.plan=plan
        #Langgraph内置的Function Call功能Agent，传入Tools即可
        self.agent_executor = create_react_agent(llm, tools, prompt=prompt_sys )
        #链式调用，提示词给大模型调用，输出结构化信息
        self.plan_executor = plan_prompt | \
                             llm2.with_structured_output(Action)

    #执行计划，在每个步骤前加了编号
    async def execute_step(self,state: PlanExecute):
        plan = state["plan"]
        plan_str = "\n".join(f"{i + 1}. {step}" for i, step in enumerate(plan))
        task = plan[0]
        task_formatted = f"""计划有以下几个步骤:
{plan_str}\n\n你需要执行 步骤{1}. {task}."""
        print(task_formatted)
        agent_response = await self.agent_executor.ainvoke(
            {"messages": [("user", task_formatted)]}
        )
        return {
            "past_steps": [(task, agent_response["messages"][-1].content)],
        }

    async def plan_step(self,state: PlanExecute):
        output = await self.plan_executor.ainvoke(state)
        if isinstance(output.actions, Response):
            return {"response": output.actions.response}
        else:
            return {"plan": output.actions.steps}

    def should_end(self,state: PlanExecute):
        if "response" in state and state["response"]:
            return END
        else:
            return "execute"

    async def run(self):
        workflow = StateGraph(PlanExecute)
        workflow.add_node("execute", self.execute_step)
        workflow.add_node("planstep", self.plan_step)

        workflow.add_edge(START, "execute")
        workflow.add_edge("execute", "planstep")

        workflow.add_conditional_edges(
            "planstep",
             self.should_end
        )
        app = workflow.compile()
        config = {"recursion_limit": 50}
        inputs = {"input": "完成所有计划后输出DONE",
                  "plan":self.plan}
        async for event in app.astream(inputs, config=config):
            for k, v in event.items():
                if k != "__end__":
                    print(v)

第一次循环

>步骤1：execute_step

计划有以下几个步骤:
1. 获取青岛啤酒的股票收盘价
2. 获取贵州茅台的股票收盘价
3. 比较青岛啤酒与贵州茅台的股票收盘价，得出哪个更贵的结论

你需要执行 步骤1. 获取青岛啤酒的股票收盘价.

>步骤2：plan_step

1.模型看到：

• 已完成：获取青岛啤酒收盘价

• 剩余计划：步骤2和步骤3

2.返回更新后的计划：['获取贵州茅台的股票收盘价', '比较青岛啤酒与贵州茅台的股票收盘价，得出哪个更贵的结论']

>步骤3：should_end

1.检查 state["response"]，没有，返回 "execute"

第二次循环

>步骤1：execute_step

计划有以下几个步骤:
1. 获取贵州茅台的股票收盘价
2. 比较青岛啤酒与贵州茅台的股票收盘价，得出哪个更贵的结论

你需要执行 步骤1. 获取贵州茅台的股票收盘价.

[
  ('获取青岛啤酒的股票收盘价', '...67.92元...'),
  ('获取贵州茅台的股票收盘价', '...1488.21元...')
]

>步骤2：plan_step（关键）

1.模型看到：

• 已完成：

  -获取青岛啤酒收盘价：67.92元

  -获取贵州茅台收盘价：1488.21元

• 剩余计划：['比较青岛啤酒与贵州茅台的股票收盘价，得出哪个更贵的结论']

2. 判断：比较步骤不需要调用工具，可直接基于已有数据得出结论。

3. 直接返回 Response：

{'response': '根据获取到的数据，贵州茅台的股票收盘价为1488.21元，而青岛啤酒的股票收盘价为67.92元。因此，贵州茅台的股价更贵。"DONE'}

>步骤3：should_end

    def should_end(self,state: PlanExecute):
        print("should_end")
        if "response" in state and state["response"]:
            return END
        else:
            return "execute"

为什么没有进入第三步？

原因：plan_step 中的模型判断“比较”步骤不需要工具调用，可直接基于已有数据给出答案，因此返回了 Response 而非 Plan，流程提前结束。

代码逻辑

    async def plan_step(self,state: PlanExecute):
        print("plan_step")
        output = await self.plan_executor.ainvoke(state)
        if isinstance(output.actions, Response):
            return {"response": output.actions.response}
        else:
            return {"plan": output.actions.steps}

提示词的影响

plan_prompt = ChatPromptTemplate.from_template(
    """ 
你是一个任务执行助手,根据给定的任务按步骤执行,不要添加任何多余的步骤 
最后一步的结果应该是最终答案。确保每个步骤都有所需的所有信息，不要跳过步骤

你的目标或问题:
{input}

你的计划是:
{plan}

当前已经完成的步骤:
{past_steps}

#请遵循以下规则
1.请输出JSON字符串,
2.相应地更新你的计划。如果不需要再执行更多步骤,就直接回复用户。否则，请填写计划。
3.只在计划中添加仍然需要完成的步骤。不要将之前完成的步骤作为计划的一部分返回。

#输出格式说明：
- 如果所有步骤都完成了，输出：{{"actions": {{"response": "最终回答内容"}}}}
- 如果还有步骤需要执行，输出：{{"actions": {{"steps": ["步骤1", "步骤2", ...]}}}}

注意：actions 字段是必需的，不要使用其他字段名。
""")

提示词要求：如果不需要再执行更多步骤，就直接回复用户。模型认为比较步骤不需要工具调用，因此直接返回最终答案。

总结

• 模型在 plan_step 中判断“比较”步骤不需要工具，可直接给出答案。

• 返回 Response 而非 Plan

• should_end 检测到 response 存在，返回 END

• 因此没有进入第三步的 execute_step

这是设计行为：模型可以智能判断是否还需要执行步骤，避免不必要的工具调用。如果需要强制执行第三步，可以在提示词中明确要求，或调整判断逻辑。

流程：

START → plan_node → execute_node → tool_node → execute_node → ... → END

执行流程：

1. plan_node：生成计划（字符串）。

2. execute_node：执行计划，可能调用工具。

3. tool_node：执行工具调用。

4. 循环：execute_node ↔ tool_node，直到出现 "Final Answer"。

流程：

START → execute → planstep → execute → planstep → ... → END

执行流程：

1. execute：执行计划中的第一步（plan[0]）。

2. planstep：更新计划，移除已完成步骤。

3. 循环：execute ↔ planstep，直到所有步骤完成。

planmode-sample：

class PlanState(MessagesState):
    plan: str

planmode-advanced：

class PlanExecute(TypedDict):
    input: str
    plan: List[str]
    past_steps: Annotated[List[Tuple], operator.add]
    response: str

planmode-sample：

def plan_node(state: PlanState):
    # 调用 LLM
    response = llm_Tongyi.invoke([SystemMessage(content=PLAN_PROMPT),state["messages"][0]])

    state["plan"] = response.content
    print("计划：\n" + state["plan"])
    return state

planmode-advanced：

    async def plan_step(self,state: PlanExecute):
        output = await self.plan_executor.ainvoke(state)
        if isinstance(output.actions, Response):
            return {"response": output.actions.response}
        else:
            return {"plan": output.actions.steps}

planmode-sample：

def execute_node(state: PlanState):
    messages = [
        SystemMessage(
            content=PLAN_EXECUTE_PROMPT.format(plan=state["plan"])
        )
    ] + state["messages"]

    response = llm_with_tools.invoke(messages)
    print("执行：\n" + response.content)
    state["messages"].append(response)

    return state

planmode-advanced：

    async def execute_step(self,state: PlanExecute):
        plan = state["plan"]
        plan_str = "\n".join(f"{i + 1}. {step}" for i, step in enumerate(plan))
        task = plan[0]
        task_formatted = f"""计划有以下几个步骤:
{plan_str}\n\n你需要执行 步骤{1}. {task}."""
        print(task_formatted)
        agent_response = await self.agent_executor.ainvoke(
            {"messages": [("user", task_formatted)]}
        )
        return {
            "past_steps": [(task, agent_response["messages"][-1].content)],
        }

planmode-sample：

def tool_node(state: PlanState):
    for tool_call in state["messages"][-1].tool_calls:
        print(tool_call)
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        state["messages"].append(ToolMessage(content=observation, tool_call_id=tool_call["id"]))
    return state

planmode-sample：

def should_continue(state: PlanState) -> Literal["tool_node", "END"]:
    messages = state["messages"]
    last_message = messages[-1]
    if "Final Answer" in last_message.content:
        return "END"
    return "tool_node"

planmode-advanced：

    def should_end(self,state: PlanExecute):
        if "response" in state and state["response"]:
            return END
        else:
            return "execute"

planmode-sample：

• 同步执行

planmode-advanced：

• 异步执行（async/await），适合并发场景。

**反思模式（Reflection）**引入独立的"反思 Agent"对主 Agent 的输出进行检查和优化，形成多智能体协作。

生成命令 → 反思检查 → 优化生成 → (最多3次迭代) → 最终输出

stop_sign=["安全隐患","木马","攻击"]
class AgentState(TypedDict):
    user_query: str #用户提问
    best_command: str #当前最优方案
    reflection: str #反思记录
    iterations: int  #执行次数

llm_Tongyi=Tongyi()

def generate_command(state: AgentState):
    iter=state["iterations"]
    print(f"生成第{iter+1}次命令")
    if iter==0: # 第一次生成
        prompt = COMMAND_PROMPT.format(
            user_query=state["user_query"],
            best_command="无",
            reflection="无"
        )
    else:
        prompt = REFLECTION_PROMPT.format(
            user_query=state["user_query"],
            best_command=state["best_command"],
            reflection=state["reflection"]
        )

    response = llm_Tongyi.invoke(prompt)
    content = response.content
    command = content.split("命令：")[1].strip()
    # 提示词中 请按以下格式输出：命令：<生成的命令>
 # 所以这里使用split("命令：")[1].strip() 获取生成的命令
    iterations = state["iterations"] + 1
    return {"best_command": command,"iterations":iterations}

def reflect_and_optimize(state: AgentState):
    print("执行反思检查")

    prompt=REFLECTION_PROMPT.format(
        command=state["best_command"],
        user_query=state["user_query"]
    )
    response = llm_Tongyi.invoke(prompt)
    content = response.content
    if "无建议" in content  or  "无需优化" in content:
        return {"reflection": "已经最优,无需优化", }
    reflection=content.split("检查结果：")[1].strip()
    return {"reflection": reflection, }

def check_reflection(state):
    if "无建议" in state["reflection"]  or  "无需优化" in state["reflection"]:
        print("已经最优,无需优化")
        return END
    for stop in stop_sign: # 如果反思结果中包含停止标志，则结束
        if stop in state["reflection"]:
            print("检测到停止标志，结束")
            return END
    if state["iterations"] >= 3:  # 最多三次迭代就结束
        print("迭代次数已达上限，结束")
        return END

    return "generate"

workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("generate", generate_command)
workflow.add_node("reflect", reflect_and_optimize)

# 设置边
workflow.set_entry_point("generate")
workflow.add_edge("generate", "reflect")

workflow.add_conditional_edges(  #条件
    "reflect",
    check_reflection
)

graph = workflow.compile()

ret = graph.invoke({"user_query": "使用docker创建nginx容器,端口映射8080:80",
        "best_command": "",
        "reflection":"",
        "iterations":0})

print("命令结果:",ret["best_command"])
print("反思结果:",ret["reflection"])

1. 核心组件与状态管理

• AgentState : 定义了在 LangGraph 节点间传递的状态数据。

• user_query : 用户的原始需求。

• best_command : 当前生成的最优 Linux 命令。

• reflection : 对当前命令的反思和优化建议。

• iterations : 记录当前的迭代次数，防止无限循环。

2. 工作节点 (Nodes)

工作流包含两个主要节点，分别代表“生成”和“反思”两个阶段：

1. generate_command (生成节点) :

• 逻辑 :

  -第一次执行时，使用 COMMAND_PROMPT 根据用户需求生成初始命令。

  -后续迭代中，结合之前的 best_command 和 reflection (优化建议)，使用 REFLECTION_PROMPT (此处代码逻辑上其实是根据反思结果重新生成，具体见 prompts.py )。

• 输出 : 更新 best_command 并增加迭代次数。

2. reflect_and_optimize (反思节点) :

• 逻辑 :

  -将生成的命令发送给模型，要求其从安全隐患、POSIX标准、效率等方面进行自我检查。

  -如果模型认为无需优化，输出“无建议”或“无需优化”。

• 输出 : 更新 reflection 字段。

3. 控制流与条件边 (Edges)

LangGraph 通过定义节点间的连接来控制执行流：

• 入口 : 流程从 generate 节点开始。

  -固定边 : generate 完成后，总是会进入 reflect 节点进行检查。

  -条件边 ( check_reflection ) : 这是 Reflection 模式的关键，它决定了流程的走向：

• 结束 (END) :

  -如果反思结果为“无建议”或“无需优化”。

  -如果检测到安全隐患（如“木马”、“攻击”等关键词）。

  -如果迭代次数达到上限（3 次）。

• 重新生成 (generate) :

  -如果反思结果包含改进建议且未达到终止条件，则流回 generate 节点，利用反思建议重新生成更好的命令。

4. 总结

这段代码通过 LangGraph 的有向图结构，清晰地表达了 “生成 -> 检查 -> (不合格则重新生成) -> 结束” 的循环优化逻辑。这种模式非常适合需要高准确性、安全性的任务（如生成运维命令、编写代码或撰写文档）。

• 高可靠性要求：运维命令、代码生成、法律文档

• 安全性敏感：需要检查安全隐患（如 rm -rf 类命令）

• 质量优化：内容创作、技术文档撰写

通过设计特殊的 ask_user 工具，在 Agent 执行过程中暂停工作流，向用户提问并等待输入，然后恢复执行。

class HumanState(MessagesState):
    query: str

async def run_graph():
    async def llm_node(state: HumanState):
        messages = [
            SystemMessage(content="你是一个仓库管理员,根据用户的要求回答相关的价格和库存信息"),
            HumanMessage(content=state["query"]),
        ] + state["messages"]

        response = await llm_with_toools.ainvoke(messages)

        state["messages"].append(response)

        return state

    async def human_node(state: HumanState):
        tool_call_id = state["messages"][-1].tool_calls[0]["id"]
        # 终止 graph执行
        content = interrupt(state["messages"][-1].tool_calls[0]["args"])
        print("用户输入是:",content)

        tool_message = ToolMessage(
            tool_call_id=tool_call_id, 
            content=content
        )

        state["messages"].append(tool_message)

        return state

    async def tool_node(state: HumanState):
        ret = []
        for tool_call in state['messages'][-1].tool_calls:
            tool_name = tool_call['name']

            get_tool = tool_with_names[tool_name]
            print("调用工具：",tool_name,tool_call['args'])
            call_tool_ret= await get_tool.ainvoke(tool_call['args'])

            state["messages"].append(ToolMessage(content=call_tool_ret,
                                tool_call_id=tool_call['id']))

        return state

    def enter_tools(state):
        if state['messages'][-1].tool_calls:
            tool_call = state['messages'][-1].tool_calls[0]
            tool_name = tool_call['name']
            print("进入工具：",tool_name)
            print("参数为：",tool_call['args'])
            if tool_name=="ask_user":
                return "human_node"
            return "tool_node"
        return END

    graph = StateGraph(HumanState)

    graph.add_node("llm_node", llm_node)
    graph.add_node("human_node", human_node)
    graph.add_node("tool_node", tool_node)

    graph.add_edge(START, "llm_node")
    graph.add_conditional_edges("llm_node", enter_tools)
    graph.add_edge("tool_node", "llm_node")
    graph.add_edge("human_node", "llm_node")

    #Checkpoint保存在保留图状态的线程中，并且可以在图执行完成后访问。 这允许图执行在特定点暂停，等待人类批准，然后从最后一个checkpoint恢复执行
    memory = MemorySaver()
    graph = graph.compile(checkpointer=memory)

    thread_config = {"configurable": {"thread_id": "123"}}

    #第一次启动
    ret = await graph.ainvoke({"query": "我想买一些苹果，总共需要多少钱"},config=thread_config)
# 如果最后一次消息是工具调用，并且工具名称为ask_user，则需要等待用户输入
    if ret['messages'][-1].tool_calls and ret['messages'][-1].tool_calls[0]['name']=="ask_user":
        sys.stdin  = io.TextIOWrapper(sys.stdin.buffer,  encoding='utf-8') # 将标准输入转换为utf-8编码
        get_user_input=input("请输入用户输入：") # 等待用户输入
        ret= await graph.ainvoke(Command( resume=get_user_input),config=thread_config) # 继续执行图

    return ret['messages'][-1].content # 返回最后一次消息的内容

1. 核心组件与状态定义

• HumanState : 继承自 MessagesState ，用于存储对话历史 ( messages ) 和用户查询 ( query )。

• MemorySaver : 检查点管理器。它将工作流的状态保存在内存中，使得工作流可以在 interrupt 暂停后，通过 thread_id 恢复到之前的状态。

2. 工作节点 (Nodes)

• llm_node (推理节点) :

  -构造系统提示词（仓库管理员身份）。

  -调用绑定了工具（ get_stock , get_price , ask_user ）的大模型。

  -将模型的回复存入 messages。

  -核心动作 : 调用 interrupt() 。

  -效果 : 此时工作流会立即暂停并保存状态，返回到调用方（即 main 函数）。

  -恢复后 : 当用户提供输入并恢复执行时， interrupt() 会返回该输入值，节点随后将其封装为 ToolMessage 并反馈给模型。

 -遍历模型发出的工具调用（如 get_stock 或 get_price ）。

 -自动执行这些函数并返回结果。

3. 控制流与条件路由

• enter_tools : llm_node 后的条件边：

  -如果模型没有调用工具 -> 结束 ( END )。

  -如果模型调用了 ask_user -> 进入 human_node (触发暂停)。

  -如果模型调用了其他工具 -> 进入 tool_node (自动执行)。

4. 暂停与恢复的执行过程

• 初次启动 :

  -graph.ainvoke({"query": "我想买一些苹果..."}) 运行。

  -LLM 决定需要询问用户“买多少”，于是调用 ask_user 工具。

  -工作流进入 human_node ，执行到 interrupt() 时 强行暂停 ， ainvoke 函数返回。

•外部干预 :

  -在 main 函数中，代码检测到最后一次动作是 ask_user 。

  -通过 input() 阻塞等待真实人类输入（例如：输入 "10斤"）。

•恢复执行 :

  -调用 graph.ainvoke(Command(resume=get_user_input), ...) 。

  -依靠 thread_id 匹配之前的状态，工作流从 human_node 的 interrupt 处 原地复活 。

  -获取到 "10斤" 后，流回 llm_node ，模型现在有了库存信息和数量需求，可以给出最终报价。

5. 总结

这段代码展示了如何处理需要“动态交互”的场景：

• 自动化 : 查询价格和库存是自动的。

• 人性化 : 当信息不足（如不知道购买数量）时，通过 interrupt 优雅地“挂起”程序，等待人类补全关键信息后再继续运行。

任务是否需要人类介入？
├── 是 → 人机协作模式
└── 否 → 任务是否要求极高可靠性？
    ├── 是 → 反思模式
    └── 否 → 工具是否可预定义？
        ├── 是 → 步骤是否固定？
        │   ├── 是 → 计划模式（简单）
        │   └── 否 → 计划模式（高级）
        └── 否 → 是否需要复杂计算？
            ├── 是 → CodeAct 模式
            └── 否 → ReAct 模式

1. 计划 + 反思：在计划模式的每个执行步骤后嵌入反思节点，确保每步输出质量

2. CodeAct + 人机协作：代码执行前请求人类确认高危操作

3. ReAct + 反思：基础工具调用后进行结果验证

Agent 设计模式的选择应基于任务复杂度、可靠性要求和成本约束三个维度权衡：

• ReAct 是通用基础，适合大多数工具调用场景

• CodeAct 释放 LLM 的编程能力，适合计算密集型任务

• 计划模式提供结构化控制，适合业务流程明确的场景

• 反思模式通过自我纠错提升质量，适合高风险场景

• 人机协作保留人类最终决策权，适合模糊需求场景

理解每种模式的设计哲学，有助于构建更强大、更可靠的 Agent 系统。

参考资源

• ReAct: Reasoning and Acting in Language Models (Yao et al., 2022)

• CodeAct: Executable Code Actions Elicit Better LLM Agents (Wang et al., 2024)

• Reflection Pattern: Self-Refine: Iterative Refinement with Self-Feedback (Madaan et al., 2023)

• Andrew Ng's Agentic Design Patterns (2024)