Agent设计模式：从单次调用到多步推理的编排架构与记忆管理

一、单次LLM调用的能力天花板：为什么Agent需要多步推理

当前大多数AI应用仍停留在"单次调用"模式——用户提问，LLM一次性生成回答。这种模式在简单问答场景下表现良好，但面对复杂任务时暴露出根本性缺陷：LLM无法使用工具、无法获取实时信息、无法自我纠错。例如，当用户问"帮我查一下北京明天的天气并推荐穿搭"时，单次调用的LLM只能基于训练数据给出泛化的回答，无法获取实时天气数据。

更深层的问题是推理深度的限制。复杂任务通常需要分解为多个子任务，每个子任务可能需要不同的工具和信息源。一个数据分析任务可能需要：先查询数据库获取原始数据，再调用统计函数计算指标，最后生成可视化图表和文字解读。单次LLM调用无法编排这种多步骤的工作流。

Agent的核心价值在于将LLM从"被动回答者"升级为"主动执行者"——它能够根据目标自主规划步骤、选择工具、执行操作、评估结果，并在必要时调整策略。这种自主性使Agent能够处理开放式的复杂任务，而非仅限于预定义的问答对。

二、ReAct循环：推理与行动的交替编排

ReAct（Reasoning + Acting）是当前最主流的Agent设计模式。其核心思想是让LLM在每一步都先"思考"（推理当前应该做什么），然后"行动"（调用工具执行），再"观察"（获取工具返回的结果），循环往复直到任务完成。

flowchart TB
    START[用户任务输入] --> THINK1[Think: 分析任务，规划步骤]
    THINK1 --> ACT1[Act: 选择并调用工具]
    ACT1 --> OBS1[Observe: 获取工具返回结果]
    OBS1 --> CHECK1{任务是否完成?}
    
    CHECK1 -->|否| THINK2[Think: 根据观察结果调整策略]
    THINK2 --> ACT2[Act: 选择并调用工具]
    ACT2 --> OBS2[Observe: 获取工具返回结果]
    OBS2 --> CHECK2{任务是否完成?}
    
    CHECK2 -->|否| THINK3[Think: 继续推理...]
    CHECK2 -->|是| RESPOND[生成最终回答]
    CHECK1 -->|是| RESPOND
    
    RESPOND --> VERIFY{回答是否合理?}
    VERIFY -->|否| THINK1
    VERIFY -->|是| OUTPUT[输出结果]

    style THINK1 fill:#e3f2fd
    style THINK2 fill:#e3f2fd
    style THINK3 fill:#e3f2fd
    style ACT1 fill:#fff3e0
    style ACT2 fill:#fff3e0
    style OBS1 fill:#e8f5e9
    style OBS2 fill:#e8f5e9
    style VERIFY fill:#ffebee

ReAct循环的关键设计决策包括：最大循环次数限制（防止无限循环）、工具选择策略（如何从工具列表中选择最合适的工具）、以及终止条件判断（何时认为任务已完成）。这些决策直接影响Agent的可靠性和效率。

三、ReAct Agent的Python实现

# react_agent.py — 基于ReAct模式的Agent核心实现
import json
import time
from dataclasses import dataclass, field
from typing import Callable, Optional
from abc import ABC, abstractmethod


@dataclass
class ToolDefinition:
    """工具定义：描述Agent可调用的工具"""
    name: str
    description: str
    parameters: dict          # JSON Schema格式的参数定义
    execute: Callable         # 工具的执行函数


@dataclass
class AgentStep:
    """Agent执行的一步记录"""
    step_number: int
    thought: str              # LLM的推理过程
    action: Optional[str]     # 选择的工具名称
    action_input: Optional[dict]  # 工具输入参数
    observation: Optional[str]    # 工具返回结果
    timestamp: float = field(default_factory=time.time)


class ReActAgent:
    """ReAct模式Agent：推理-行动-观察循环"""

    def __init__(
        self,
        llm_call: Callable,           # LLM调用函数
        tools: list[ToolDefinition],   # 可用工具列表
        max_steps: int = 8,            # 最大循环次数
        verbose: bool = False,
    ):
        self.llm_call = llm_call
        self.tools = {t.name: t for t in tools}
        self.max_steps = max_steps
        self.verbose = verbose
        self._history: list[AgentStep] = []

    def run(self, task: str) -> str:
        """执行Agent任务，返回最终回答"""
        system_prompt = self._build_system_prompt()
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": task},
        ]

        for step_num in range(1, self.max_steps + 1):
            # 调用LLM进行推理
            llm_response = self.llm_call(messages)
            
            # 解析LLM输出：提取Thought、Action、Action Input
            parsed = self._parse_response(llm_response)

            if self.verbose:
                print(f"[Step {step_num}] Thought: {parsed['thought']}")

            # 如果LLM认为任务完成，直接返回
            if parsed.get("finish"):
                return parsed["finish"]

            # 执行工具调用
            tool_name = parsed.get("action")
            tool_input = parsed.get("action_input", {})

            if tool_name not in self.tools:
                observation = f"错误：工具 '{tool_name}' 不存在。可用工具: {list(self.tools.keys())}"
            else:
                try:
                    tool = self.tools[tool_name]
                    result = tool.execute(**tool_input)
                    observation = json.dumps(result, ensure_ascii=False)
                except Exception as e:
                    observation = f"工具执行失败: {type(e).__name__}: {e}"

            if self.verbose:
                print(f"[Step {step_num}] Action: {tool_name}({tool_input})")
                print(f"[Step {step_num}] Observation: {observation[:200]}")

            # 记录步骤
            step = AgentStep(
                step_number=step_num,
                thought=parsed.get("thought", ""),
                action=tool_name,
                action_input=tool_input,
                observation=observation,
            )
            self._history.append(step)

            # 将观察结果追加到消息中，继续循环
            messages.append({"role": "assistant", "content": llm_response})
            messages.append({
                "role": "user",
                "content": f"Observation: {observation}\n\n请继续推理，或输出 Final Answer: 完成任务。"
            })

        # 超过最大步数，强制总结
        messages.append({
            "role": "user",
            "content": "已达到最大推理步数。请基于已有观察结果，给出最终回答。"
        })
        return self.llm_call(messages)

    def _build_system_prompt(self) -> str:
        """构建系统提示词：包含工具描述和ReAct格式要求"""
        tool_descriptions = []
        for name, tool in self.tools.items():
            params_desc = json.dumps(tool.parameters, ensure_ascii=False)
            tool_descriptions.append(f"- {name}: {tool.description}\n  参数: {params_desc}")

        tools_text = "\n".join(tool_descriptions)

        return f"""你是一个智能助手，能够通过推理和工具调用来完成用户任务。

可用工具:
{tools_text}

请严格按照以下格式回复:

Thought: 分析当前情况，思考下一步应该做什么
Action: 要调用的工具名称
Action Input: 工具的输入参数（JSON格式）

当你认为任务已完成时，使用以下格式:
Thought: 任务已完成
Final Answer: 最终回答内容

重要规则:
1. 每次只能调用一个工具
2. 必须基于观察结果进行推理，不要猜测工具的返回值
3. 如果工具执行失败，尝试其他方法或调整参数"""

    def _parse_response(self, response: str) -> dict:
        """解析LLM的回复，提取Thought、Action等字段"""
        result = {}

        # 提取Thought
        if "Thought:" in response:
            thought_start = response.index("Thought:") + len("Thought:")
            thought_end = len(response)
            for marker in ["Action:", "Final Answer:"]:
                if marker in response[thought_start:]:
                    pos = response.index(marker, thought_start)
                    thought_end = min(thought_end, pos)
            result["thought"] = response[thought_start:thought_end].strip()

        # 检查是否完成
        if "Final Answer:" in response:
            fa_start = response.index("Final Answer:") + len("Final Answer:")
            result["finish"] = response[fa_start:].strip()
            return result

        # 提取Action
        if "Action:" in response:
            action_start = response.index("Action:") + len("Action:")
            action_end = len(response)
            if "Action Input:" in response[action_start:]:
                action_end = response.index("Action Input:", action_start)
            result["action"] = response[action_start:action_end].strip()

        # 提取Action Input
        if "Action Input:" in response:
            input_start = response.index("Action Input:") + len("Action Input:")
            input_text = response[input_start:].strip()
            try:
                result["action_input"] = json.loads(input_text)
            except json.JSONDecodeError:
                result["action_input"] = {"raw_input": input_text}

        return result

    @property
    def history(self) -> list[AgentStep]:
        """获取执行历史"""
        return self._history

上述实现中，ReActAgent 通过系统提示词引导LLM按照"Thought-Action-Observation"的格式输出，解析器从LLM的文本输出中提取结构化信息。每一步的工具执行结果作为新的观察追加到对话中，驱动下一轮推理。最大步数限制防止了无限循环。

四、ReAct模式的局限性与架构权衡

Token消耗：ReAct循环中，每一步的Thought、Action和Observation都消耗Token。一个8步的Agent任务，Token消耗可能是单次调用的5到8倍。对于成本敏感的场景，需要权衡推理深度与成本预算。优化方向是压缩历史步骤的上下文——超过4步的历史可以摘要为一段简短的回顾，而非原样保留。

可靠性瓶颈：Agent的可靠性取决于LLM正确选择工具和生成合法参数的能力。当工具数量超过10个时，LLM的工具选择准确率会显著下降。缓解方案是将工具按功能域分组，每次只向LLM暴露与当前任务相关的工具子集。

调试困难：Agent的执行路径不可预测，同一任务可能每次走不同的步骤。必须记录完整的执行历史（包括每步的Thought），否则无法复现和调试异常行为。生产环境中，建议将Agent的每一步执行都写入持久化存储，支持事后回放和分析。

五、总结

ReAct模式为Agent提供了从单次调用到多步推理的编排框架。通过"思考-行动-观察"的循环，Agent能够自主规划、工具调用和策略调整。在落地时，需要关注Token消耗、工具选择准确率和执行可观测性三个关键问题。建议从3到5个工具的小规模场景起步，逐步扩展工具集和推理深度，同时建立完善的执行日志和回放机制。