智能体架构¶

本章介绍 LLM 智能体的核心架构模式。我们以 Hermes Agent 作为具体案例，但这些模式广泛应用于 AutoGen、CrewAI、LangGraph 等框架中。

系统概览¶

┌─────────────────────────────────────────────────────────────┐
│                   用户界面层                                 │
│        (终端、电报、Discord、Slack、Web)                       │
└─────────────────────────┬───────────────────────────────────┘
                          │
                          ▼
┌─────────────────────────────────────────────────────────────┐
│                    智能体核心层                              │
│    ┌─────────────────────────────────────────────────────┐  │
│    │           智能体循环 (感知 → 规划 → 执行)              │  │
│    │   构建提示词 → 调用 LLM → 处理响应                      │  │
│    └─────────────────────────────────────────────────────┘  │
│                          │                                   │
│         ┌────────────────┼────────────────┐                  │
│         ▼                ▼                ▼                  │
│  ┌────────────┐  ┌────────────┐  ┌────────────┐             │
│  │   工具     │  │   记忆     │  │   委托     │             │
│  │ (68+ 内置) │  │(持久 + 会话)│  │  (子智能体) │             │
│  └────────────┘  └────────────┘  └────────────┘             │
└─────────────────────────────────────────────────────────────┘

1. 智能体循环：五种设计模式¶

**智能体循环**是核心执行引擎。最新研究表明有五种主流模式：

模式 1: ReAct（推理 + 行动）¶

将推理过程与工具执行交叉进行，是目前使用最广泛的范式。

# ReAct 循环：思考 → 行动 → 观察 → 重复
messages = [{"role": "user", "content": user_input}]

for i in range(max_turns):
    # 1. 调用 LLM
    response = llm.invoke(messages)

    # 2. 解析响应
    if response.tool_calls:
        for tc in response.tool_calls:
            # 3. 执行工具
            result = tools[tc.name](**tc.arguments)
            # 4. 作为观察结果追加
            messages.append({
                "role": "tool",
                "content": f"观察结果：{result}",
                "tool_call_id": tc.id,
            })
    else:
        return response.content  # 完成

模式 2: 先规划后执行¶

先制定完整计划，再逐步执行。相比 ReAct，长链路任务稳定性更高。

def plan_and_execute(task: str, llm, tools):
    # 阶段 1: 规划
    plan_response = llm.invoke(f"""
        将此任务分解为有序步骤：
        任务：{task}
        返回编号步骤列表。
    """)
    steps = parse_steps(plan_response)

    # 阶段 2: 按顺序执行
    results = []
    for step in steps:
        result = llm.invoke(f"执行步骤：{step}\n上下文：{results}")
        results.append(result)

    return summarize(results)

模式 3: 反思 / 自我优化¶

生成输出后，内部评审器评估质量并触发修订，直到达到质量阈值。

from pydantic import BaseModel, Field

class Review(BaseModel):
    score: int = Field(..., description="1-10 质量评分")
    feedback: str = Field(..., description="改进建议")

structured_llm = llm.with_structured_output(Review)

def self_refine(task: str, threshold: int = 8, max_iters: int = 5):
    draft = llm.invoke(f"完成：{task}")

    for _ in range(max_iters):
        review: Review = structured_llm.invoke(f"评分并改进：\n{draft}")
        if review.score >= threshold:
            return draft
        draft = llm.invoke(f"根据反馈改进：{review.feedback}\n原文：{draft}")

    return draft

模式 4: 监督者 + 工作者¶

一个管理智能体将任务分配给专业工作者，并行或顺序执行。

def supervisor_router(task: str, workers: list, llm):
    decision = llm.invoke(f"""
        任务：{task}
        可用工作者：{list(workers.keys())}
        选择最合适的工作者并说明原因。
    """)
    worker_name = parse_worker_choice(decision)
    return workers[worker_name].execute(task)

模式 5: 路由器 / 选择器¶

轻量级分类器将输入路由到不同模型、工具或智能体路径。

def route(task: str, llm):
    decision = llm.invoke(f"""
        将此任务分类为：code | general | research | math
        只输出分类名称。
        任务：{task}
    """).content.strip().lower()

    return {
        "code": code_agent,
        "general": general_agent,
        "research": research_agent,
        "math": math_agent,
    }.get(decision, general_agent).execute(task)

2. 工具系统¶

工具定义¶

每个工具有三个组成部分：

tool = {
    "name": "terminal",
    "description": "在终端中执行 Shell 命令。",
    "parameters": {
        "type": "object",
        "properties": {
            "command": {
                "type": "string",
                "description": "要执行的 Shell 命令。",
            }
        },
        "required": ["command"],
    },
}

LLM 在系统提示词中接收此 schema，决定**何时**以及**如何**调用工具。

工具分发¶

class ToolRegistry:
    def dispatch(self, name: str, arguments: dict) -> str:
        tool = self.tools[name]
        if tool["check_fn"] and not tool["check_fn"]():
            return "错误：工具条件不满足。"
        return tool["handler"](arguments)

# 关键原则：工具永远不抛异常——只返回结构化结果

工具分类¶

类别	工具	示例
执行	shell, python, code	运行命令、执行代码
文件 I/O	read, write, patch	读写文件
网络	search, extract, fetch	浏览网页
浏览器	navigate, click, screenshot	Web 浏览器自动化
视觉	describe, analyze, ocr	图像理解
记忆	save, load, search	持久存储
委托	spawn, delegate	创建子智能体
调度	cron, schedule	定时任务
通信	send_email, send_message	发送消息

3. 记忆系统：从 RAG 到智能体记忆¶

智能体记忆经历了三个阶段的演变，理解这个演进对构建生产智能体至关重要。

阶段 1: 经典 RAG（2020–2023）¶

检索增强生成：将文档预先索引到向量数据库，查询时检索相关片段。

离线：文档 → 分块 → 向量化 → 存入 VectorDB
在线：查询 → 向量化 → 检索 top-k 片段 → 拼接 → LLM → 响应

局限性： 只读。无法从交互中学习。"一本静态的百科全书"。

阶段 2: Agentic RAG（2023–2024）¶

RAG 成为智能体决定何时、如何使用的工具。

智能体决定：
  - 需要检索吗？从哪个来源？
  - 这个上下文相关吗？
  - 应该尝试不同的搜索策略吗？

局限性： 仍是只读。无从交互中学习。

阶段 3: 真正的智能体记忆（2024+）¶

智能体可以**创建、更新和删除**记忆——从每次交互中学习。

┌─────────────────────────────────────────────────────────────┐
│              智能体记忆架构（2024+）                          │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌─────────────────────────────────────────────────────┐   │
│  │       记忆形成 → 记忆演化 → 记忆检索                    │   │
│  └─────────────────────────────────────────────────────┘   │
│                          │                                   │
│  ┌─────────────────────────────────────────────────────┐   │
│  │  形式（记忆如何存储）                                  │   │
│  │  ┌──────────────────┬──────────────────┐              │   │
│  │  │ Token 级记忆     │ 参数化记忆        │              │   │
│  │  │ （显式、可搜索） │ （编码进模型权重）│              │   │
│  │  └──────────────────┴──────────────────┘              │   │
│  │                                                      │   │
│  │  功能（记忆用于什么）                                  │   │
│  │  ┌──────────────────┬──────────────────┐              │   │
│  │  │ 事实记忆         │ 经验记忆          │              │   │
│  │  │ （用户信息、     │ （成功案例、     │              │   │
│  │  │  环境信息）      │   技能、案例）   │              │   │
│  │  └──────────────────┴──────────────────┘              │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

Hermes Agent 中的记忆实现¶

# ~/.hermes/memories/
# ├── user.json        # 用户画像（事实型）
# ├── memory.json      # 环境事实（事实型）
# └── sessions/        # 会话记录（经验型）

def inject_memory(agent: AIAgent) -> str:
    sections = []

    profile = load_json("~/.hermes/user.json")
    if profile:
        sections.append(f"## 用户画像\n{json.dumps(profile, indent=2)}")

    facts = load_json("~/.hermes/memory.json")
    if facts:
        sections.append(f"## 记忆\n" + "\n".join(f"- {f}" for f in facts))

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

上下文压缩¶

当对话变长时，Hermes 会压缩历史：

def compress_context(messages: list, target_ratio: float = 0.2) -> list:
    """通过约 80% 压缩长对话历史"""
    protected = messages[-self.protect_last_n:]  # 保留最近上下文
    compressible = messages[:-self.protect_last_n]

    summary = self.llm.summarize(compressible)

    return [
        {"role": "system", "content": f"前文摘要：{summary}"},
        *protected,
    ]

4. 提示词构建¶

**系统提示词**由多个组件组装而成：

def build_system_prompt(agent: AIAgent) -> str:
    sections = []

    # 1. 基础人格 (SOUL.md)
    sections.append(load_file("~/.hermes/SOUL.md"))

    # 2. 记忆注入
    sections.append(inject_memory(agent))

    # 3. 相关技能
    relevant_skills = load_relevant_skills(agent.user_input)
    for skill in relevant_skills:
        sections.append(f"## 技能：{skill.name}\n{skill.content}")

    # 4. 工具 schema
    tool_schemas = [t["schema"] for t in agent.tools.values()]
    sections.append(f"## 可用工具\n{json.dumps(tool_schemas)}")

    # 5. 上下文文件
    sections.append(load_context_files())

    return "\n\n---\n\n".join(sections)

5. 安全：护栏与审批¶

由于智能体可以执行任意命令，安全检查至关重要：

命令审批¶

危险模式 = [
    r'rm\s+-rf\s+/',           # 递归删除根目录
    r'git\s+push\s+--force',   # 强制推送
    r'dd\s+if=',               # 直接磁盘写入
    r'mkfs',                   # 格式化文件系统
]

def 应该审批(command: str) -> bool:
    for pattern in 危险模式:
        if re.search(pattern, command):
            return True
    return False

输出护栏¶

from pydantic import BaseModel, Field

class GuardrailedOutput(BaseModel):
    decision: str = Field(..., description="决策或回答")
    reason: str = Field(..., description="解释")
    approved: bool = Field(..., description="是否应展示给用户")

6. Hermes Agent 作为具体实现¶

Hermes Agent 在生产级系统中实现了以上所有模式：

组件	Hermes 实现
智能体循环	`run_agent.py` — AIAgent 类
工具	`tools/` 目录中的 68+ 工具
记忆	`~/.hermes/memories/` (JSON + SQLite)
技能	`~/.hermes/skills/` (带 frontmatter 的 Markdown)
委托	`spawn` / `delegate` 工具
压缩	`agent/compression.py`
网关	`gateway/` (Telegram、Discord、Slack 等)
凭证池	多 API 密钥轮换

7. 模式对比：何时使用哪种¶

模式	最佳场景	复杂度	可靠性
ReAct	问答、网络搜索、开放域任务	低	中等
先规划后执行	长报告、复杂分解	中等	高
自我优化	代码生成、创意写作	中等	高
监督者 + 工作者	多工具流水线	中高	高
路由器	多模型成本优化	低	中等

8. 跨框架的模式迁移¶

这些架构模式在不同框架间是可迁移的：

模式	Hermes	AutoGen/MAF	CrewAI	LangGraph
智能体循环	`AIAgent`	`ConversableAgent`	`Agent`	图节点
工具	`registry.register()`	`code_execution_config`	`tools=[]`	工具节点
记忆	JSON/SQLite	自定义	内置	State dict
护栏	手动	`human_input_mode`	工具	中断节点
委托	`spawn` 工具	嵌套智能体	Crew 层级	条件边
路由	Gateway	GroupChat	Process 类型	`should_continue`

参考资料¶

ReAct 论文 — 推理 + 行动范式
AI 智能体时代的记忆：综述 — 全面记忆分类（形式-功能-动态）
构建有效智能体 (Anthropic) — 生产模式
Hermes Agent 架构 — 具体实现
LangGraph 文档 — 状态图模式
LLM 智能体常见设计模式 — 模式分类