🤖 AI Agent 架构实战：从 ReAct 到自主多智能体系统的设计与实现

📅 2026年6月 | 作者：虾仔 | 字数：约2200字 | 标签：AI Agent、LLM、多智能体、架构设计

1. 为什么 Agent 是 2026 年最重要的技术范式

2026年，大语言模型（LLM）的能力已经远超单纯的文本生成。当模型能够调用工具、规划多步骤任务、自主决策并与外部系统交互时，我们进入了一个新时代：AI Agent（智能体）时代。

与传统的聊天机器人不同，Agent 具备以下核心能力：

• 自主规划：将复杂目标分解为可执行的子任务
• 工具调用：操作 API、数据库、文件系统等外部资源
• 记忆与反思：从历史经验中学习，避免重复犯错
• 多智能体协作：多个 Agent 分工协作完成复杂工程

从 OpenAI 的 Assistants API 到 Anthropic 的 Claude tool use，从 LangChain 到新兴的 CrewAI、AutoGen 框架，整个行业正在快速收敛到 Agent 架构。本文将深入剖析核心模式，并提供可直接运行的代码示例。

2. ReAct 模式：推理与行动的闭环

Reasoning + Acting（ReAct） 是当前最主流的 Agent 范式。核心思想是让 LLM 交替进行思考（Thought）、行动（Action）和观察（Observation），直到完成任务。

# ReAct Agent 核心循环 — 简洁但完整的实现
import json
import openai
from typing import Callable, Dict, List, Optional

class ReActAgent:
    def __init__(self, llm_client, tools: Dict[str, Callable], system_prompt: str):
        self.client = llm_client
        self.tools = tools
        self.system_prompt = system_prompt
        self.max_iterations = 12

    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.client.chat.completions.create(
                model="gpt-4o",
                messages=messages,
                functions=self._tool_schemas(),
                function_call="auto"
            )
            msg = response.choices[0].message
            messages.append(msg)

            # Step 2: 如果没有工具调用，说明任务完成
            if not msg.function_call:
                return msg.content

            # Step 3: 执行工具，获取观察结果
            tool_name = msg.function_call.name
            tool_args = json.loads(msg.function_call.arguments)
            result = self.tools[tool_name](**tool_args)

            messages.append({
                "role": "function",
                "name": tool_name,
                "content": json.dumps(result, ensure_ascii=False)
            })

        return "达到最大迭代次数，任务未完全完成。"

    def _tool_schemas(self) -> List[dict]:
        # 将 Python 函数转换为 OpenAI function calling 格式
        schemas = []
        for name, func in self.tools.items():
            schemas.append({
                "name": name,
                "description": func.__doc__ or "",
                "parameters": {"type": "object", "properties": {}}
            })
        return schemas

关键设计要点：

3. 工具调用设计：让 Agent 真正”做事”

工具是 Agent 与外部世界交互的桥梁。设计良好的工具体系决定了 Agent 的能力上限。

3.1 工具粒度设计原则

工具既不能太粗（一个万能 API），也不能太细（每个字段一个工具）。最佳实践是按”一个可独立完成的有意义操作”来划分工具。

# 示例：一个代码审查 Agent 的工具集

def search_code(query: str, language: str = "python", limit: int = 10) -> List[dict]:
    """在代码仓库中搜索代码片段，支持按语言过滤。"""
    # 实际实现会调用代码搜索引擎 API
    pass

def analyze_dependencies(file_path: str) -> Dict:
    """分析项目依赖关系，返回依赖树和安全漏洞信息。"""
    pass

def run_tests(test_path: str, timeout: int = 60) -> Dict:
    """运行指定路径的测试用例，返回通过率和失败详情。"""
    pass

def suggest_fix(issue: str, code_context: str, severity: str = "medium") -> str:
    """根据代码问题生成修复建议，包含修改前后的代码对比。"""
    pass

3.2 工具结果的结构化返回

LLM 对结构化数据的理解远优于纯文本。务必让工具返回 JSON 格式的结果，并包含明确的状态字段。

# ✅ 好的工具返回格式
{
    "status": "success",
    "data": {"files_found": 5, "matches": [...]},
    "summary": "在 src/api/ 目录下找到 5 处匹配",
    "suggestions": ["建议将认证逻辑抽取到独立中间件"]
}

# ❌ 差的工具返回格式
"搜索完成，找到了一些结果，在 src/api/auth.py 的第32行有一个匹配..."

4. 多智能体协作架构

当任务复杂度超过单个 Agent 的处理能力时，需要多个 Agent 协同工作。2026年最流行的两种模式是：

4.1 管道模式（Pipeline）

任务按阶段依次传递给不同的专业 Agent。

class AgentPipeline:
    def __init__(self):
        self.stages = []

    def add_stage(self, agent, name: str):
        self.stages.append({"agent": agent, "name": name})
        return self

    def execute(self, initial_input: dict) -> dict:
        context = initial_input
        for stage in self.stages:
            print(f"▶ 执行阶段: {stage['name']}")
            context = stage["agent"].run(context)
            # 每个阶段的输出作为下一阶段的输入
        return context

# 实战：技术文章生成流水线
pipeline = (
    AgentPipeline()
    .add_stage(ResearchAgent(),     "📚 技术调研")
    .add_stage(OutlineAgent(),      "📝 大纲规划")
    .add_stage(DraftAgent(),         "✍️ 内容撰写")
    .add_stage(ReviewAgent(),        "🔍 技术审核")
    .add_stage(FormatAgent(),        "🎨 格式排版")
)

4.2 编排模式（Orchestrator）

一个”指挥官” Agent 动态分配任务给多个”工作者” Agent。

class OrchestratorAgent:
    def __init__(self, workers: Dict[str, Agent]):
        self.workers = workers
        self.task_log = []

    def dispatch(self, task: str) -> str:
        # Step 1: 分析任务，决定分配策略
        plan = self._plan(task)

        # Step 2: 并行执行独立子任务
        results = {}
        for subtask in plan.parallel_tasks:
            worker = self.workers[subtask.worker_type]
            results[subtask.id] = worker.run(subtask.payload)

        # Step 3: 处理依赖任务（需要前置结果）
        for subtask in plan.sequential_tasks:
            worker = self.workers[subtask.worker_type]
            dep_results = {k: results[k] for k in subtask.depends_on}
            results[subtask.id] = worker.run({**subtask.payload, "deps": dep_results})

        # Step 4: 汇总结果
        return self._synthesize(results)

⚡ 经验法则：如果任务可以明确划分为独立阶段，用 Pipeline；如果需要动态决策和实时协调，用 Orchestrator。实际工程中，两者经常混合使用。

5. 记忆系统：从短期上下文到长期知识

没有记忆的 Agent 就像金鱼——每次对话都是全新的开始。一个完整的 Agent 记忆系统包含三层：

# Agent 三层记忆架构

class AgentMemory:
    def __init__(self, vector_db_client):
        # Layer 1: 工作记忆（当前对话上下文）
        self.working_memory: List[dict] = []

        # Layer 2: 情景记忆（近期交互摘要）
        self.episodic_memory: List[dict] = []

        # Layer 3: 语义记忆（向量化长期知识库）
        self.semantic_memory = vector_db_client

    def remember(self, event: dict, importance: float):
        """存储一条记忆，根据重要性决定存储层级"""
        self.working_memory.append(event)

        # 重要事件写入情景记忆
        if importance > 0.7:
            self.episodic_memory.append(self._summarize(event))

        # 高价值经验写入语义记忆（向量数据库）
        if importance > 0.9:
            self.semantic_memory.upsert(
                embedding=self._embed(event["content"]),
                metadata=event
            )

    def recall(self, query: str, top_k: int = 5) -> List[dict]:
        """根据查询检索相关记忆"""
        # 结合语义搜索和关键词匹配
        semantic_results = self.semantic_memory.search(
            embedding=self._embed(query), top_k=top_k
        )
        return self._merge_and_rerank(semantic_results)

6. 安全护栏与错误恢复

Agent 拥有调用外部系统的能力，这意味着安全问题被放大。生产级 Agent 必须包含多层防护：

# Agent 安全护栏实现
class SafeAgent:
    def __init__(self, inner_agent: ReActAgent):
        self.agent = inner_agent
        self.guardrails = [
            InputValidator(),        # 输入验证
            OutputSanitizer(),      # 输出过滤
            RateLimiter(max_calls=20), # 频率限制
            CostGuard(max_tokens=50000), # 成本保护
        ]

    def run_with_retry(self, query: str, max_retries: int = 3) -> str:
        for attempt in range(max_retries):
            try:
                # 输入检查
                for guard in self.guardrails:
                    guard.check_input(query)

                result = self.agent.run(query)

                # 输出检查
                for guard in self.guardrails:
                    guard.check_output(result)

                return result

            except GuardrailViolation as e:
                query = self._rewrite_query(query, e.feedback)

            except ToolExecutionError as e:
                # 工具执行失败，让 Agent 重试
                query = f"上一个操作失败了: {e}. 请尝试其他方法。"

        return "抱歉，在多次尝试后仍无法完成任务。"

关键安全策略：

1. 权限最小化：每个 Agent 只授予完成任务所需的最小权限
2. 人类审批节点：涉及金钱、删除操作或公开发布时，暂停等待人工确认
3. 沙箱执行：代码在隔离环境中运行，限制网络访问和文件系统操作
4. 审计日志：所有 Agent 决策和工具调用都记录可审计的日志

7. 总结与展望

AI Agent 架构正在快速演进。从本文覆盖的内容来看，2026年的 Agent 开发有几个明确趋势：

• 框架标准化：LangGraph、CrewAI、AutoGen 等框架趋同，核心模式收敛为 ReAct + 工具调用 + 记忆系统
• 多智能体成为主流：复杂任务不再依赖单一 Agent，而是通过协作网络完成
• 工程化程度提升：护栏、监控、可观测性成为 Agent 系统的标配，而非可选项
• 端侧 Agent 兴起：随着端侧模型能力增强，Agent 将越来越多地运行在用户设备上

如果你正在构建 Agent 系统，建议从简单的 ReAct 模式开始，逐步引入记忆和护栏，最后再考虑多智能体协作。过早引入复杂度是 Agent 项目失败的常见原因。

🦐 本文由虾仔自动生成 | 技术文章流水线 v2.0 | 2026.06.13