AI Agent 架构设计与实战：从 ReAct 到多智能体协作

2026年，AI Agent 已经从实验室概念走向生产环境。无论是自动化运维、智能客服，还是代码生成、数据分析，Agent 都在重新定义人与 AI 的协作方式。本文将深入剖析 Agent 的核心架构，并通过完整代码示例带你从零构建一个可落地的智能体系统。

1. 什么是 AI Agent？

简单来说，AI Agent 是一个能够自主感知环境、制定计划、调用工具并迭代完成目标的系统。与传统的大模型”一问一答”不同，Agent 具备以下核心能力：

• 自主决策：根据上下文动态规划执行步骤
• 工具调用：操作外部 API、数据库、文件系统等
• 记忆能力：维护短期工作记忆和长期知识库
• 多轮迭代：根据执行结果反馈修正策略

想象一下，你告诉 Agent “帮我分析本周的销售数据并生成报告”，它会自己读取数据、运行分析、生成图表、输出文档——整个过程无需逐步指令。

2. ReAct 框架：Agent 的基石

ReAct（Reason + Act）是当前最主流的 Agent 框架，核心思想是让模型交替进行”思考”和”行动”：

• Thought：模型分析当前状态，决定下一步
• Action：执行具体工具调用
• Observation：获取执行结果，反馈给模型

下面是一个精简但完整的 ReAct Agent 实现：

from openai import OpenAI
import json

TOOLS = {
    "web_search": lambda q: f"[搜索结果] 关于 '{q}' 的模拟搜索结果...",
    "calculator": lambda expr: str(eval(expr)),
    "get_weather": lambda city: f"{city}: 晴, 22°C, 湿度 65%",
}

TOOL_SCHEMAS = [
    {"name": "web_search", "description": "搜索互联网获取信息", "parameters": {"query": "string"}},
    {"name": "calculator", "description": "执行数学计算", "parameters": {"expression": "string"}},
    {"name": "get_weather", "description": "查询城市天气", "parameters": {"city": "string"}},
]

SYSTEM_PROMPT = """你是一个具备工具调用能力的 AI Agent。
当需要调用工具时，严格按以下格式回复：
Action: {tool_name}
Action Input: {input_json}

当已获得足够信息回答时，按以下格式回复：
Final Answer: {你的回答}"""

def run_agent(user_query: str, max_rounds: int = 8):
    client = OpenAI()
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": user_query}
    ]

    for round_i in range(max_rounds):
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=[{"type": "function", "function": {
                "name": t["name"],
                "description": t["description"],
                "parameters": {
                    "type": "object",
                    "properties": {k: {"type": "string"} for k in t["parameters"].values()},
                    "required": list(t["parameters"].keys())
                }
            }} for t in TOOL_SCHEMAS]
        )

        msg = response.choices[0].message
        messages.append(msg)

        if msg.tool_calls:
            for call in msg.tool_calls:
                func_name = call.function.name
                args = json.loads(call.function.arguments)
                # 合并参数为简单输入
                args_str = " ".join(args.values()) if len(args) == 1 else json.dumps(args)
                result = TOOLS.get(func_name, lambda _: "未知工具")(args_str)
                messages.append({
                    "role": "tool",
                    "tool_call_id": call.id,
                    "content": result
                })
                print(f"[Round {round_i+1}] {func_name}({args_str}) → {result}")
        else:
            print(f"\n✅ 最终回答:\n{msg.content}")
            return msg.content

# 运行示例
run_agent("北京今天天气怎么样？气温适合穿什么？")

这个示例展示了 ReAct 的完整循环：模型思考 → 调用工具 → 观察结果 → 再次思考，直到得出最终答案。

3. 记忆系统：让 Agent 拥有”记忆力”

没有记忆的 Agent 每次对话都是”失忆状态”。一个实用的记忆系统通常包含两层：

from collections import deque
import time

class AgentMemory:
    def __init__(self, max_short=20, max_long=100):
        # 短期记忆：最近 N 轮对话
        self.short_term = deque(maxlen=max_short)
        # 长期记忆：重要信息持久化
        self.long_term = []
        self.max_long = max_long

    def add_short(self, role: str, content: str):
        self.short_term.append({
            "role": role,
            "content": content,
            "timestamp": time.time()
        })

    def add_long(self, key: str, value: str, importance: float = 0.8):
        """存储重要信息到长期记忆，importance 越高越不容易被淘汰"""
        self.long_term.append({
            "key": key,
            "value": value,
            "importance": importance,
            "timestamp": time.time()
        })
        # 超过容量时淘汰低重要度记忆
        if len(self.long_term) > self.max_long:
            self.long_term.sort(key=lambda x: x["importance"])
            self.long_term.pop(0)

    def get_short_context(self):
        return [{"role": m["role"], "content": m["content"]} for m in self.short_term]

    def search_long(self, query: str, top_k=3):
        """简单的关键词匹配，生产环境建议用向量搜索"""
        scored = []
        query_words = set(query.lower().split())
        for item in self.long_term:
            overlap = len(query_words & set(item["value"].lower().split()))
            score = overlap * item["importance"]
            scored.append((score, item))
        scored.sort(key=lambda x: -x[0])
        return [item for _, item in scored[:top_k]]

# 使用示例
memory = AgentMemory()
memory.add_short("user", "帮我分析销售数据")
memory.add_short("assistant", "好的，正在处理...")
memory.add_long("user_pref", "用户偏好使用柱状图展示数据", 0.9)
memory.add_long("data_source", "销售数据位于 /data/sales_2026.csv", 0.95)

print("短期记忆:", len(memory.get_short_context()), "条")
print("长期记忆搜索:", memory.search_long("数据图表"))

生产环境中，长期记忆应使用向量数据库（如 Chroma、Pinecone）替代关键词匹配，配合 Embedding 模型实现语义搜索。

4. 多智能体协作：1+1>2

单个 Agent 能力有限，复杂任务需要多智能体协作。常见模式包括：

• 管道模式：A 的输出作为 B 的输入
• 分工模式：不同 Agent 负责不同子任务
• 辩论模式：多个 Agent 讨论后达成共识

以下是一个”研究员 + 编辑”双 Agent 协作系统：

class ResearchAgent:
    """负责信息搜集和分析"""
    def run(self, topic: str):
        print(f"🔍 研究员正在分析: {topic}")
        # 模拟深度研究过程
        return {
            "topic": topic,
            "findings": [
                f"{topic} 市场规模年增长率达 35%",
                f"头部玩家占据了 60% 市场份额",
                f"新兴技术正在重塑行业格局"
            ],
            "data": {"growth_rate": 0.35, "top3_share": 0.60}
        }

class EditorAgent:
    """负责将研究内容整理为成品报告"""
    def run(self, research_data: dict) -> str:
        print(f"✍️ 编辑正在整理报告: {research_data['topic']}")
        report = f"# {research_data['topic']} 深度报告\n\n"
        report += "## 核心发现\n"
        for i, finding in enumerate(research_data["findings"], 1):
            report += f"{i}. {finding}\n"
        report += f"\n## 关键指标\n"
        report += f"- 年增长率: {research_data['data']['growth_rate']*100}%\n"
        report += f"- Top3 市占率: {research_data['data']['top3_share']*100}%\n"
        return report

class Pipeline:
    """编排多 Agent 协作流程"""
    def __init__(self):
        self.agents = []

    def add_agent(self, agent):
        self.agents.append(agent)
        return self

    def execute(self, initial_input):
        data = initial_input
        for agent in self.agents:
            data = agent.run(data)
        return data

# 运行多 Agent 协作
pipeline = Pipeline()
pipeline.add_agent(ResearchAgent())
pipeline.add_agent(EditorAgent())

report = pipeline.execute("2026年AI Agent市场趋势")
print("\n" + "="*50)
print(report)

这个管道模式可以轻松扩展——加入审核 Agent、翻译 Agent、图表生成 Agent 等，构建完整的内容生产线。

5. 生产环境落地建议

将 Agent 从 Demo 推向生产，需要关注以下几个关键点：

• 错误恢复：工具调用可能失败，Agent 需要优雅处理异常并重试。建议设置最大重试次数和降级策略。
• 安全防护：Agent 拥有工具调用能力，必须做好权限控制。敏感操作（删除、支付、外发）需要人工确认。
• 成本控制：每次工具调用都消耗 Token，合理设置 max_rounds 上限，避免 Agent 陷入无限循环。
• 可观测性：记录每轮 Thought/Action/Observation，方便调试和优化。LangSmith、Langfuse 等工具是不错的选择。
• 提示词工程：Agent 的行为高度依赖 System Prompt。建议用结构化的方式描述工具、约束和输出格式。

总结

AI Agent 正在从”能聊天的 AI”进化为”能干活 AI”。ReAct 框架提供了坚实的理论基础，记忆系统赋予 Agent 持续服务能力，多智能体协作则解锁了复杂场景的处理能力。

2026 年的 Agent 生态已经相当成熟——从 LangChain、LlamaIndex 这样的开发框架，到 AutoGPT、CrewAI 这样的编排平台，再到 Dify、Coze 这样的低代码工具，开发者可以根据需求灵活选择。

核心建议：从小场景切入，先让 Agent 在一个具体任务上跑通，再逐步扩展能力边界。 不要试图一步到位造一个全能 Agent，而是构建一个能真正解决痛点的专用智能体。