🤖 Agent 架构实战指南

§1 Agent 是什么？为什么需要专门的架构？

Agent 不是一个简单的 LLM 封装。它是一个自主决策系统——接收目标，规划步骤，调用工具，处理异常，持续迭代，直到目标达成。

与普通 LLM 调用的本质区别：

§2 架构决策树：我该选什么架构？

§3 六大核心架构模式

从 20 个项目中提炼出的六种架构模式，覆盖从简单到复杂的全部场景：

3.1 ReAct 循环 — 最小可行架构

ReAct（Reasoning + Acting）是 Agent 的"Hello World"。9/20 项目以此为核心。

谁做得最好：

• Smolagents — 最简洁的 ReAct 实现，while not is_done: 一目了然
• Agent Zero — 在 ReAct 基础上加了 LLM 后过滤和自进化
• SWE-agent — 8 种输出解析器，解决 ReAct 最大的痛点（LLM 输出格式不稳定）

Python 实现（Smolagents 风格）：

class ReactAgent:
    def __init__(self, model, tools, max_steps=10):
        self.model = model
        self.tools = {t.name: t for t in tools}
        self.max_steps = max_steps

    def run(self, task: str) -> str:
        messages = [{"role": "user", "content": task}]

        for step in range(self.max_steps):
            # Think: LLM 决定下一步
            response = self.model.chat(messages)
            messages.append({"role": "assistant", "content": response})

            # 解析：提取动作和参数
            action = self.parse_action(response)
            if action is None:  # 没有动作 = 完成
                return response

            # Act: 执行工具
            tool_name, tool_input = action
            if tool_name not in self.tools:
                observation = f"Error: Tool '{tool_name}' not found"
            else:
                try:
                    observation = self.tools[tool_name].run(tool_input)
                except Exception as e:
                    observation = f"Error: {e}"

            # Observe: 把结果喂回 LLM
            messages.append({
                "role": "user",
                "content": f"Observation: {observation}"
            })

        return "Max steps reached"

# 使用
agent = ReactAgent(
    model=LiteLLM("gpt-4o"),
    tools=[WebSearch(), FileRead(), CodeRun()],
)
result = agent.run("查找 Python 3.13 的新特性并写一个摘要")

3.2 Plan-Execute — 结构化任务的正确选择

先规划，再执行。适合步骤相对固定的场景，如数据处理流水线、代码生成。

谁做得最好：

• GPT Researcher — Planner 生成子问题 → 并行搜索 → 综合报告。66KB 深度分析
• Aider — Architect-Coder 双模型：Architect 规划，Coder 执行

# Plan-Execute 模式
class PlanExecuteAgent:
    def __init__(self, planner_model, executor_model, tools):
        self.planner = planner_model
        self.executor = executor_model
        self.tools = tools

    def run(self, task: str) -> str:
        # Phase 1: Plan
        plan = self.planner.chat(
            f"将以下任务分解为具体步骤：\n{task}\n"
            "输出 JSON 格式: {\"steps\": [...]}"
        )
        steps = json.loads(plan)["steps"]

        # Phase 2: Execute
        results = []
        for i, step in enumerate(steps):
            result = self.executor.run(step, self.tools)
            results.append(result)

            # Phase 3: Replan if needed
            if self.needs_replan(result, steps[i:]):
                updated_plan = self.planner.chat(
                    f"步骤 {i} 的结果与预期不符：{result}\n"
                    f"剩余步骤：{steps[i+1:]}\n"
                    "请更新计划："
                )
                steps = json.loads(updated_plan)["steps"]

        return "\n".join(results)

3.3 状态机 — 复杂流程的精确控制

用显式状态转移替代隐式的 LLM 决策。适合需要严格流程控制的企业场景。

谁做得最好：

• LangGraph — 最成熟的状态机 Agent 框架。Pregel 执行引擎 + Checkpoint/Time Travel
• PydanticAI — 类型安全的状态机，GraphAgent

# LangGraph 风格的状态机
from langgraph.graph import StateGraph, END

class AgentState(TypedDict):
    messages: list
    next_action: str
    retry_count: int

def analyze(state):
    # 分析用户输入
    result = llm.invoke(state["messages"])
    return {"next_action": result.action, "messages": [result]}

def plan(state):
    # 生成计划
    plan = llm.invoke(f"制定计划：{state['messages'][-1]}")
    return {"messages": [plan]}

def execute(state):
    # 执行工具
    result = tool_executor.run(state["messages"][-1].tool_calls)
    return {"messages": [result], "retry_count": 0}

def should_plan(state):
    if state["next_action"] == "complex":
        return "plan"
    return "execute"

# 构建图
graph = StateGraph(AgentState)
graph.add_node("analyze", analyze)
graph.add_node("plan", plan)
graph.add_node("execute", execute)

graph.add_conditional_edges("analyze", should_plan)
graph.add_edge("plan", "execute")
graph.add_edge("execute", END)

app = graph.compile()

3.4 Actor Model — 多 Agent 对话的正确姿势

每个 Agent 是独立的 Actor，通过消息传递通信。适合需要 Agent 间协商、辩论的场景。

谁做得最好：

• AutoGen — Actor Model 的集大成者。5 种群组编排模式 + SocietyOfMind 嵌套 + Handoff 工具调用建模
• CAMEL — Role-Playing 对话框架，Agent 间自动协商角色

# AutoGen 风格的 Actor Model
from autogen import ConversableAgent, GroupChat, GroupChatManager

researcher = ConversableAgent(
    "researcher",
    system_message="你是研究员，负责收集信息",
    llm_config={"model": "gpt-4o"},
    human_input_mode="NEVER",
)

writer = ConversableAgent(
    "writer",
    system_message="你是写作者，负责根据研究结果写文章",
    llm_config={"model": "gpt-4o"},
    human_input_mode="NEVER",
)

reviewer = ConversableAgent(
    "reviewer",
    system_message="你是审稿人，负责检查文章质量",
    llm_config={"model": "gpt-4o"},
    human_input_mode="NEVER",  # 或 "TERMINATE" 需要人工终审
)

# 群组对话
group_chat = GroupChat(
    agents=[researcher, writer, reviewer],
    messages=[],
    max_round=10,
    speaker_selection_method="round_robin",  # 或 "auto" (LLM选择)
)

manager = GroupChatManager(groupchat=group_chat)
researcher.initiate_chat(
    manager,
    message="请研究 AI Agent 架构的最新进展并写一篇综述",
)

3.5 Orchestrator-Worker — SOP 驱动的协作

一个中心编排器分配任务给专业 Worker。适合流水线式的复杂任务。

谁做得最好：

• MetaGPT — SOP Watch-Publish 编排，最接近真实软件团队的协作模式
• CrewAI — Process-driven 编排 + Flow 装饰器，简洁的 API
• AutoGen MagenticOne — Orchestrator 循环 + 专门化 Worker

# MetaGPT 风格的 SOP 编排
from metagpt.roles import Role
from metagpt.schema import Message

class ProductManager(Role):
    name: str = "Alice"
    profile: str = "Product Manager"

    async def _act(self) -> Message:
        # 生成 PRD
        prd = await self._write_prd()
        return Message(content=prd, role=self.profile)

class Architect(Role):
    name: str = "Bob"
    profile: str = "Architect"

    async def _act(self) -> Message:
        # 根据PRD设计系统架构
        design = await self._design_system()
        return Message(content=design, role=self.profile)

class Engineer(Role):
    name: str = "Charlie"
    profile: str = "Engineer"

    async def _act(self) -> Message:
        # 根据设计写代码
        code = await self._write_code()
        return Message(content=code, role=self.profile)

# SOP：PM → Architect → Engineer
async def software_sop(requirement: str):
    pm = ProductManager()
    architect = Architect()
    engineer = Engineer()

    prd_msg = await pm.run(requirement)
    design_msg = await architect.run(prd_msg)
    code_msg = await engineer.run(design_msg)
    return code_msg

3.6 自进化架构 — Agent 的终极形态

Agent 不仅执行任务，还能反思自己的表现、修改自身行为。这是目前最前沿的方向。

谁做得最好：

• Hermes Agent — Background Review Fork 是唯一真正的自进化设计：后台 fork 一个 Review Agent，审计主 Agent 的行为，发现可优化点后修改 prompt/工具配置
• Agent Zero — LLM Infection Check（防止 prompt 注入改变自身）+ 3 区域记忆

§4 工具系统设计

工具是 Agent 的"手和脚"。工具系统的设计直接决定 Agent 的能力边界。

4.1 工具定义的三个流派

{
  "name": "web_search",
  "description": "搜索互联网",
  "parameters": {
    "type": "object",
    "properties": {
      "query": {"type": "string"},
      "max_results": {"type": "integer"}
    },
    "required": ["query"]
  }
}

from pydantic import BaseModel

class SearchArgs(BaseModel):
    query: str
    max_results: int = 5

@agent.tool
async def web_search(
    ctx: RunContext,
    args: SearchArgs,
) -> str:
    results = await search(args.query)
    return results[:args.max_results]

# search.yaml
name: search
command: ./scripts/search.sh
args:
  - name: query
    type: string
  - name: max_results
    type: integer
    default: 5

4.2 工具安全：三级审批机制

不同工具有不同的风险等级。好的工具系统不是一刀切，而是分级的：

最佳实践组合（参考 安全机制）：

# OpenClaw 风格的 Policy Pipeline
class ToolPolicy:
    def check(self, tool: str, args: dict) -> PolicyResult:
        tool_def = self.registry.get(tool)

        # 1. 静态规则检查
        if tool_def.risk_level == "RED":
            if not self.user_approved(tool, args):
                return PolicyResult(deny=True, reason="需要用户审批")

        # 2. 动态上下文检查（Hermes 风格）
        threats = self.threat_scanner.scan(args)
        if threats:
            return PolicyResult(deny=True, reason=f"检测到威胁: {threats}")

        # 3. 沙箱执行（SWE-agent/OpenHands 风格）
        if tool_def.requires_sandbox:
            return PolicyResult(sandbox=True)

        return PolicyResult(allow=True)

# Codex CLI 风格的三级 Profile
class ApprovalProfile:
    SUGGEST = "suggest"   # 建议但不阻止
    AUTO_EDIT = "auto-edit"  # 文件编辑自动通过
    FULL_AUTO = "full-auto"  # 全自动（危险！）

4.3 工具发现与推荐

当工具数量超过 10 个，LLM 选错工具的概率急剧上升。两个解决方案：

• BM25 + LLM 混合推荐（MetaGPT）— 先 BM25 检索候选，再 LLM 精排
• 语义缓存（Hermes Experience Pool）— 记住"什么任务用什么工具"

§5 记忆与上下文管理

这是 Agent 架构中最被低估的模块。上下文窗口有限，如何高效利用是核心挑战。

5.1 五级记忆架构

从无状态到自进化，记忆系统的 5 个层级：

5.2 Context Condenser — 上下文压缩的艺术

OpenHands 的 LLMSummarizingCondenser 是目前最优雅的解决方案：

# OpenHands 风格的 Condenser
class Condenser:
    """上下文压缩器——把无限历史压缩进有限窗口"""

    def condense(self, events: list[Event], max_tokens: int) -> list[Event]:
        current_tokens = sum(e.token_count for e in events)

        if current_tokens <= max_tokens:
            return events  # 不需要压缩

        # 策略 1: 丢弃旧的观察结果（保留动作）
        events = self._drop_old_observations(events)
        if self._token_count(events) <= max_tokens:
            return events

        # 策略 2: 摘要压缩早期历史
        early, recent = self._split(events, ratio=0.4)
        summary = self.llm.summarize(early)
        return [SummaryEvent(summary)] + recent

    def _drop_old_observations(self, events):
        """只保留最近 N 次工具返回的完整结果，旧的结果截断"""
        obs_count = 0
        result = []
        for e in reversed(events):
            if e.type == "observation":
                obs_count += 1
                if obs_count > 5:  # 只保留最近5次
                    e = e.truncate(max_chars=200)
            result.append(e)
        return list(reversed(result))

5.3 Aider 的 RepoMap — 代码记忆的最佳实践

当 Agent 操作代码仓库时，如何把几千个文件塞进有限的上下文？Aider 的 RepoMap 用 PageRank 算法解决：

# Aider RepoMap 简化原理
class RepoMap:
    def get_repo_map(self, chat_files: list[str]) -> str:
        # 1. 解析所有文件的 AST，提取定义和引用
        tags = self._get_tags()  # {filepath: [(name, kind, line), ...]}

        # 2. 构建引用图：谁引用了谁
        G = self._build_graph(tags)

        # 3. PageRank 排序——被引用最多的定义最重要
        rankings = pagerank(G)

        # 4. 只保留 Top-K 重要的定义，用压缩格式输出
        top_tags = sorted(tags, key=lambda t: rankings[t], reverse=True)
        map_str = ""
        for tag in top_tags:
            if len(map_str) > MAX_MAP_TOKENS:
                break
            map_str += f"{tag.filepath}:{tag.line} {tag.name}\n"

        return map_str  # 几千文件 → 几百行摘要

§6 提示词工程：组装与缓存

Agent 的 prompt 不是写一次就完事——它是动态组装的，每次调用可能不同。

6.1 Prompt 组装三层次（Hermes 风格）

# Hermes 风格的三层 Prompt 组装
class PromptComposer:
    def compose(self, agent, user_input, memory, tools):
        parts = []

        # Layer 1: Identity (静态，可缓存)
        parts.append(self._identity_block(agent))
        parts.append(self._capabilities_block(agent))

        # Layer 2: Context (动态，每轮变化)
        parts.append(self._memory_block(memory))
        parts.append(self._tools_block(tools))
        parts.append(self._user_input_block(user_input))

        # Layer 3: Safety (静态，可缓存)
        parts.append(self._safety_guardrails())
        parts.append(self._output_format_spec())

        return "\n\n".join(parts)

6.2 Cache Boundary — 省 90% Token 的秘密（OpenClaw）

OpenClaw 的 Cache Boundary 是提示词缓存的教科书级实现：

# OpenClaw Cache Boundary 原理
#
# LLM API 的 Prompt Cache 机制：
# - 前缀匹配的 prompt 只计算一次
# - 后续请求如果前缀不变，只计算新增部分
#
# 关键：把静态内容放在前面，动态内容放在后面！

COMPOSED_PROMPT = """
[SYSTEM - 静态，可缓存]          ← Cache Boundary
├── 角色定义                      ← 几乎不变
├── 工具描述                      ← 偶尔变
├── 安全约束                      ← 几乎不变
│
[CACHE BOUNDARY]                  ← 在此插入标记
│
[USER - 动态，不缓存]
├── 记忆检索结果                  ← 每轮变化
├── 当前对话                      ← 每轮变化
└── 用户最新输入                  ← 每轮变化
"""

# 实现效果：
# 第1轮: 计算全部 4000 tokens
# 第2轮: 只计算新增 ~500 tokens (前缀命中缓存)
# 节省: ~87% token 计算成本

§7 错误处理与容错

Agent 运行在不确定的环境中——LLM 输出不稳定、工具调用失败、网络超时。容错设计是生产系统的心脏。

7.1 四层容错架构

# 综合：生产级容错 Agent
class ProductionAgent:
    def __init__(self, ...):
        self.retry_policy = RetryPolicy(max_retries=3, backoff="exponential")
        self.stuck_detector = StuckDetector(
            patterns=["repeat_action", "no_progress", "error_loop"],
            window_size=5,
        )
        self.checkpointer = FileCheckpointer("./checkpoints/")

    @checkpoint  # 自动保存检查点
    def run(self, task: str) -> str:
        for step in range(self.max_steps):
            # 卡住检测
            if self.stuck_detector.is_stuck(self.history):
                self.stuck_detector.intervene(self)  # 注入反思 prompt
                continue

            # 带重试的执行
            result = self.retry_policy.execute(
                lambda: self._step(task),
                retry_on=[OutputParseError, ToolTimeoutError],
            )

            if result.is_final:
                return result.output

        return "任务未完成（达到步数上限）"

§8 多 Agent 协作模式对比

当你确定需要多 Agent 时，选择哪种协作模式？

§9 完整架构蓝图

把以上所有模块组合起来，一个生产级 Agent 系统的完整架构：

9.1 最小可行 Agent（3 个模块）

如果你想最快上线一个 Agent：

# 最小可行 Agent: ReAct + Function Calling + 对话历史
from litellm import completion

TOOLS = [
    {"type": "function", "function": {
        "name": "web_search", "description": "搜索互联网",
        "parameters": {"type": "object", "properties": {
            "query": {"type": "string"}}, "required": ["query"]}
    }}
]

def run_agent(task: str, max_steps: int = 5) -> str:
    messages = [{"role": "system", "content": "你是一个有用的助手。"}, 
                {"role": "user", "content": task}]

    for _ in range(max_steps):
        resp = completion(model="gpt-4o", messages=messages, tools=TOOLS)
        msg = resp.choices[0].message
        messages.append(msg)

        if msg.tool_calls is None:
            return msg.content  # Agent 完成

        # 执行工具
        for tc in msg.tool_calls:
            result = execute_tool(tc.function.name, 
                                 json.loads(tc.function.arguments))
            messages.append({"role": "tool", 
                           "tool_call_id": tc.id, 
                           "content": str(result)})

    return "达到最大步数"

# 就这么简单。3 个函数，50 行代码。

9.2 生产级 Agent（7 个模块）

加上安全、缓存、容错、记忆：

# 生产级 Agent 配置
class ProductionAgentConfig:
    # 核心循环
    loop_type: str = "react"  # react | plan_execute | state_machine

    # LLM 配置
    model: str = "gpt-4o"
    fallback_model: str = "claude-sonnet"  # Failover
    cache_boundary: bool = True  # 提示词缓存

    # 工具系统
    tools: list[Tool] = []
    tool_policy: PolicyPipeline = PolicyPipeline()
    sandbox: bool = True  # 工具在沙箱中执行

    # 记忆系统
    memory_level: int = 2  # 0-4, 推荐 L2
    condenser: Condenser = LLMSummarizingCondenser(max_tokens=4000)

    # 安全
    human_approval_for: list[str] = ["RED"]  # 需要人工审批的风险级别
    threat_scanner: Scanner = ContextThreatScanner()
    audit_log: bool = True

    # 容错
    max_retries: int = 3
    stuck_detector: StuckDetector = StuckDetector()
    checkpointer: Checkpointer = FileCheckpointer()

    # 成本控制
    max_tokens_per_run: int = 100_000
    max_cost_per_run: float = 1.0  # USD

§10 框架选型指南

如果你不想从零构建，选择哪个框架？

§11 反模式与常见坑

§12 与其他模块的关联

Agent 架构不是孤立的，它与 SaaS 全栈知识库的其他模块紧密关联：

§13 Agent 构建模块拆分（深度链接）

本文是从 20 个 Agent 项目研究中提炼的概览。每个子模块有独立深度分析页：

§14 关键数据：20 个项目架构统计

§15 趋势与展望

趋势 1：混合架构 — ReAct 内核 + 外层编排

单一 ReAct 已经不够。趋势是 ReAct 作为核心执行单元，外层用状态机/Orchestrator 编排多个 ReAct Agent。LangGraph 的 Pregel 引擎就是这个思路。

趋势 2：ACI > HCI — 为 Agent 设计专用接口

SWE-agent 的核心洞察：不要让 Agent 复用人类接口，而是设计 Agent-Computer Interface (ACI)。Shell 命令的输出太冗长？设计精简的 Agent 专用命令。文件浏览太慢？提供 grep + 定位的一体化接口。

趋势 3：A2A 协议 — Agent 间通信标准化

Google ADK 的 A2A (Agent-to-Agent) 协议：让不同框架的 Agent 可以互相通信。就像 HTTP 统一了 Web 通信，A2A 可能统一 Agent 通信。

趋势 4：Background Review — 持续自优化

Hermes 的 Background Review Fork 是目前唯一的实现，但方向清晰：Agent 在后台持续审计自己的行为，发现可优化点并自动调整。这是 Agent 从"工具"到"同事"的关键一步。