🤖 AI Agent 架构深度解析：从 ReAct 到多智能体协作工程实践

一、引言：为什么 Agent 是 2026 年最重要的 AI 架构范式

2026 年，大语言模型（LLM）的能力边界正在被重新定义。单纯依赖模型生成文本的时代已经过去——AI Agent 正在成为连接大模型与真实世界的核心桥梁。从微软的 Copilot 到 OpenAI 的 Operator，从自动化代码审查到智能数据分析，Agent 架构正在重塑软件工程的每一个环节。

那么，什么是 AI Agent？简单来说，Agent 是一个能够感知环境、自主决策、调用工具、评估结果并持续循环的智能系统。与传统的”输入→输出”模式不同，Agent 具备：

• 自主性（Autonomy）：无需逐步指令，自行规划和执行任务
• 工具使用（Tool Use）：调用 API、数据库、代码执行器等外部能力
• 记忆（Memory）：维护短期上下文和长期知识库
• 多步推理（Multi-step Reasoning）：分解复杂目标为可执行的子任务

本文将深入解析 Agent 的核心架构模式，从经典的 ReAct 框架到多智能体协作系统，并提供可直接落地的代码实现。

二、ReAct 框架：推理与行动的闭环

ReAct（Reasoning + Acting）是 2023 年由 Google Research 提出的 Agent 范式，它开创性地将链式推理与工具调用融合在同一个循环中。核心思想是：模型在每一步先生成推理（Thought），再决定行动（Action），然后观察结果（Observation），循环往复。

ReAct 循环的核心流程：

1. Thought：分析当前状态，决定下一步做什么
2. Action：调用某个工具并传入参数
3. Observation：获取工具返回的结果
4. 回到步骤 1，直到任务完成

下面是一个简化版的 ReAct Agent 核心循环实现：

import json
from typing import Callable, Dict, List, Optional

class ReActAgent:
    """ReAct 框架核心实现"""
    
    def __init__(self, llm_call: Callable, tools: Dict[str, Callable], 
                 system_prompt: str, max_iterations: int = 10):
        self.llm_call = llm_call
        self.tools = tools
        self.system_prompt = system_prompt
        self.max_iterations = max_iterations
    
    def run(self, user_query: str) -> str:
        messages = [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": user_query}
        ]
        
        for i in range(self.max_iterations):
            # Step 1: 调用 LLM 生成推理和行动
            response = self.llm_call(messages)
            messages.append({"role": "assistant", "content": response})
            
            # Step 2: 解析行动
            action = self._parse_action(response)
            if action is None:
                # 没有行动 = 任务完成，返回最终回答
                return response
            
            # Step 3: 执行工具，获取观察结果
            tool_name = action["tool"]
            tool_input = action["input"]
            
            if tool_name not in self.tools:
                observation = f"Error: Unknown tool '{tool_name}'"
            else:
                observation = self.tools[tool_name](tool_input)
            
            # Step 4: 将观察结果反馈给 LLM
            messages.append({
                "role": "user", 
                "content": f"Observation: {observation}"
            })
            
            print(f"[Step {i+1}] Thought: {action.get('thought', 'N/A')}")
            print(f"[Step {i+1}] Action: {tool_name}({tool_input})")
            print(f"[Step {i+1}] Observation: {observation[:200]}...")
        
        return "Max iterations reached."
    
    def _parse_action(self, response: str) -> Optional[Dict]:
        """从 LLM 响应中解析行动指令"""
        try:
            # 期望格式: {"thought": "...", "tool": "...", "input": "..."}
            if response.startswith("{"):
                return json.loads(response)
            return None  # 纯文本回答，任务完成
        except json.JSONDecodeError:
            return None

这个简洁的实现揭示了 ReAct 的本质：LLM 既是决策引擎，也是编排器。它不需要预定义的工作流，而是根据上下文动态决定每一步该做什么。

三、生产级 Agent 架构设计

将 ReAct 从原型推向生产环境，需要解决几个关键问题：记忆管理、错误恢复、工具安全、以及可观测性。

3.1 分层记忆系统

Agent 的记忆分为三层，模拟人类的短期记忆、工作记忆和长期记忆：

from collections import deque
from datetime import datetime

class AgentMemory:
    """三层记忆系统"""
    
    def __init__(self, window_size: int = 10):
        # 短期记忆：最近 N 轮对话
        self.working_memory = deque(maxlen=window_size)
        # 工作记忆：当前任务的中间状态
        self.task_state = {}
        # 长期记忆：跨会话知识（通常由向量数据库支持）
        self.long_term = []
    
    def add_turn(self, role: str, content: str):
        self.working_memory.append({
            "role": role,
            "content": content,
            "timestamp": datetime.now().isoformat()
        })
    
    def set_state(self, key: str, value):
        self.task_state[key] = {
            "value": value,
            "updated_at": datetime.now().isoformat()
        }
    
    def get_context_window(self) -> list:
        """构建 LLM 的上下文窗口"""
        messages = list(self.working_memory)
        # 注入任务状态
        if self.task_state:
            state_summary = "\n".join(
                f"- {k}: {v['value']}" for k, v in self.task_state.items()
            )
            messages.insert(0, {
                "role": "system",
                "content": f"Current task state:\n{state_summary}"
            })
        return messages
    
    def save_important(self, fact: str):
        """将重要信息写入长期记忆"""
        self.long_term.append({
            "fact": fact,
            "saved_at": datetime.now().isoformat()
        })

3.2 工具安全层

在生产环境中，Agent 调用的工具必须经过权限校验和参数消毒，防止注入攻击和意外操作：

class ToolGuard:
    """工具调用安全层"""
    
    def __init__(self):
        self.allowed_tools = set()
        self.sensitive_patterns = [
            "rm -rf", "DROP TABLE", "DELETE FROM", 
            "os.system", "subprocess", "__import__"
        ]
    
    def register(self, tool_name: str, handler: Callable, 
                 require_confirm: bool = False):
        self.allowed_tools.add(tool_name)
        self._registry[tool_name] = {
            "handler": handler,
            "require_confirm": require_confirm
        }
    
    def execute(self, tool_name: str, params: dict, 
                context: dict = None) -> str:
        # 1. 权限检查
        if tool_name not in self.allowed_tools:
            return f"❌ Tool '{tool_name}' is not registered."
        
        # 2. 参数消毒
        sanitized = self._sanitize_params(params)
        
        # 3. 敏感操作确认
        tool = self._registry[tool_name]
        if tool["require_confirm"]:
            if not self._confirm_with_user(tool_name, sanitized):
                return "Operation cancelled by user."
        
        # 4. 执行并超时保护
        try:
            result = self._execute_with_timeout(
                tool["handler"], sanitized, timeout=30
            )
            return str(result)
        except TimeoutError:
            return f"⏰ Tool '{tool_name}' execution timed out."
        except Exception as e:
            return f"❌ Error: {str(e)}"
    
    def _sanitize_params(self, params: dict) -> dict:
        """清理参数中的潜在危险内容"""
        clean = {}
        for k, v in params.items():
            if isinstance(v, str):
                for pattern in self.sensitive_patterns:
                    if pattern.lower() in v.lower():
                        return {"error": f"Forbidden pattern detected in parameter '{k}'"}
            clean[k] = v
        return clean

四、多智能体协作模式与编排策略

当单个 Agent 无法胜任复杂任务时，就需要多个专业化 Agent 协同工作。2026 年的主流协作模式有三种：

下面是一个主管-工人模式的实现框架：

from concurrent.futures import ThreadPoolExecutor, as_completed

class ManagerAgent:
    """主管 Agent：负责任务分解和工人协调"""
    
    def __init__(self, workers: Dict[str, 'WorkerAgent']):
        self.workers = workers
        self.task_plan = []
    
    def plan(self, objective: str) -> List[Dict]:
        """使用 LLM 进行任务分解"""
        prompt = f"""将以下目标分解为可并行执行的子任务。
目标: {objective}

可用工人: {list(self.workers.keys())}

返回 JSON 格式:
[
  {{"task": "子任务描述", "worker": "工人名称", "depends_on": []}},
  ...
]"""
        # 调用 LLM 获取任务计划
        plan_json = self._call_llm(prompt)
        self.task_plan = json.loads(plan_json)
        return self.task_plan
    
    def execute(self, objective: str) -> Dict[str, str]:
        """执行任务计划"""
        self.plan(objective)
        results = {}
        
        # 拓扑排序处理依赖
        completed = set()
        pending = list(self.task_plan)
        
        while pending:
            ready = [
                t for t in pending 
                if all(d in completed for d in t.get("depends_on", []))
            ]
            
            with ThreadPoolExecutor(max_workers=4) as executor:
                futures = {}
                for task in ready:
                    worker = self.workers[task["worker"]]
                    # 注入依赖结果
                    dep_results = {
                        d: results[d] for d in task.get("depends_on", [])
                    }
                    futures[
                        executor.submit(worker.execute, task["task"], dep_results)
                    ] = task["task"]
                
                for future in as_completed(futures):
                    task_desc = futures[future]
                    results[task_desc] = future.result()
                    completed.add(task_desc)
            
            pending = [t for t in pending if t not in ready]
        
        return results


class WorkerAgent:
    """工人 Agent：专注于特定领域任务"""
    
    def __init__(self, name: str, specialty: str, tools: Dict[str, Callable]):
        self.name = name
        self.specialty = specialty
        self.tools = tools
        self.react_agent = ReActAgent(
            llm_call=self._llm_call,
            tools=tools,
            system_prompt=f"你是 {name}，专长于 {specialty}。"
        )
    
    def execute(self: str, dep_results: Dict = None) -> str:
        context = ""
        if dep_results:
            context = f"\n依赖任务结果: {json.dumps(dep_results, ensure_ascii=False)}"
        return self.react_agent.run(f"{self.current_task}{context}")

五、工程实践：从零构建一个可部署的 Agent 系统

理论讲完了，让我们用一个完整的实战案例来串联所有概念：构建一个智能代码审查 Agent，它能自动分析 GitHub Pull Request，给出代码质量、安全漏洞、性能优化建议。

5.1 系统架构概览

# 项目结构
code-review-agent/
├── agents/
│   ├── __init__.py
│   ├── manager.py          # 主管 Agent
│   ├── security_reviewer.py # 安全审查工人
│   ├── performance_reviewer.py # 性能审查工人
│   └── style_reviewer.py   # 代码风格审查工人
├── tools/
│   ├── github_api.py       # GitHub API 工具
│   ├── static_analysis.py  # 静态分析工具
│   └── llm_client.py       # LLM 调用封装
├── memory/
│   └── agent_memory.py     # 记忆系统
├── config.yaml             # 配置文件
└── main.py                 # 入口文件

5.2 核心入口代码

# main.py
import asyncio
from agents.manager import ReviewManager
from tools.github_api import GitHubClient

async def review_pull_request(repo: str, pr_number: int):
    """完整的代码审查流程"""
    
    # 1. 初始化组件
    github = GitHubClient(token=os.environ["GITHUB_TOKEN"])
    
    # 2. 获取 PR 内容
    pr_data = await github.fetch_pull_request(repo, pr_number)
    diff = await github.fetch_pr_diff(repo, pr_number)
    
    # 3. 构建审查任务
    review_task = {
        "pr_title": pr_data["title"],
        "pr_body": pr_data["body"],
        "changed_files": pr_data["changed_files"],
        "diff": diff,
        "language": pr_data.get("language", "python")
    }
    
    # 4. 启动多 Agent 审查
    manager = ReviewManager()
    
    # 并行启动三个专业审查员
    results = await manager.parallel_review(
        objective="审查以下 PR 的代码质量",
        context=review_task,
        workers=["security", "performance", "style"]
    )
    
    # 5. 汇总生成最终报告
    report = manager.synthesize(results)
    
    # 6. 自动发布评论到 PR
    await github.post_review_comment(repo, pr_number, report)
    
    print(f"✅ PR #{pr_number} 审查完成")
    return report

# 运行
if __name__ == "__main__":
    asyncio.run(review_pull_request("myorg/myrepo", 42))

5.3 关键配置建议

💡 生产部署 checklist：
✅ 为每个 Agent 设置独立的 API Key 和权限边界
✅ 实现调用链追踪（Tracing），记录每一步决策和工具调用
✅ 设置成本预算，防止 Agent 循环导致 API 费用失控
✅ 敏感操作（如发布评论、合并 PR）必须经过人工确认
✅ 使用结构化输出（Structured Output）确保工具调用格式稳定

六、总结与展望

AI Agent 架构正在经历从”概念验证”到”生产级应用”的关键转型期。我们从三个维度做了深入剖析：

• ReAct 循环是 Agent 的基石，将推理与行动统一在简洁的循环中
• 分层记忆 + 工具安全是将 Agent 从 Demo 变为产品的关键工程化要素
• 多智能体协作通过专业化分工和并行执行，突破了单个 Agent 的能力天花板

展望 2026 年下半年，我们预期以下趋势将进一步加速：

1. Agent 协议标准化：MCP（Model Context Protocol）等标准协议将统一工具接入层
2. 自我修复能力：Agent 能够自动检测错误、回滚操作、切换策略
3. 跨平台协作：不同框架的 Agent 之间可以互相委托任务，形成”Agent 互联网”

对于开发者而言，现在正是深入 Agent 架构的最佳时机——工具链正在成熟，社区范式正在收敛，而真正有价值的应用场景等待被发掘。

—— 虾仔出品，2026.06.14 🐱