# Agent 源码级设计模式参考

> 基于 25 个开源项目的源代码分析，提取可复用的设计模式、数据结构、算法

---

## 一、Agent Loop 设计模式

### 1.1 Smolagents — ReAct 循环（最简洁实现）

来源: `src/smolagents/agents.py` (~1000 行)

```python
# 核心循环（_run_stream）
while not returned_final_answer and step_number <= max_steps:
    # 可选: planning step（每 N 步）
    if planning_interval and step_number % planning_interval == 0:
        plan = model.generate(planning_prompt)
        memory.steps.append(PlanningStep(plan=plan))

    # 必选: action step
    action = ActionStep(step_number=step_number)
    output = self._step_stream(action)  # 子类实现
    if output.is_final_answer:
        returned_final_answer = True
    memory.steps.append(action)
```

**关键数据结构**:
```python
class AgentMemory:
    system_prompt: SystemPromptStep
    steps: list[TaskStep | PlanningStep | ActionStep | FinalAnswerStep]

class ActionStep:
    model_input_messages: list[ChatMessage]   # 模型输入
    model_output_message: ChatMessage          # 原始输出
    tool_calls: list[ToolCall]                 # 解析后的工具调用
    observations: str                          # 工具执行结果
    action_output: Any                         # 最终输出
    error: AgentError | None
```

### 1.2 Grok Build — run_session() 循环

来源: `xai-grok-shell/src/session/acp_session_impl/run_loop.rs`

```
run_session():
  while not finished and turn < max_turns:
    1. prompt_build.rs    → 组装上下文（system + memory + tools + history）
    2. Sampler            → LLM 推理（HTTP SSE 流式）
    3. tool_dispatch.rs   → 解析 tool_calls → 权限门控 → 执行
    4. 结果注入上下文 → loop
    5. 上下文压缩检查（>85% → compact）
```

**关键差异**: Grok Build 在 Rust 中实现，使用 Actor 模式管理终端和子代理，比 Python 版本多了沙箱集成和 Doom Loop 检测。

### 1.3 通用 Agent Loop 骨架（可直接复用）

```python
class AgentLoop:
    def run(self, task: str) -> FinalAnswer:
        state = State(task=task)
        memory = Memory()

        for turn in range(max_turns):
            # 1. Context Builder
            context = self._build_context(state, memory, turn)

            # 2. LLM Call
            response = self.llm.chat(context, tools=tools_schema)

            # 3. Parse
            if response.is_final:
                return FinalAnswer(response.text)

            # 4. Execute Tools
            for tc in response.tool_calls:
                result = self._execute(tc, state)
                state.add_observation(tc.name, result)

            # 5. Loop Control
            if state.should_stop():
                break

        return FinalAnswer("max turns reached")
```

---

## 二、工具系统设计模式

### 2.1 Grok Build — NewTool trait + 依赖注入

```rust
// Rust trait（接口契约）
pub trait NewTool {
    fn name(&self) -> &str;
    fn description(&self) -> &str;
    fn parameters(&self) -> Vec<ToolParam>;
    fn execute(&self, args: Value) -> Result<String>;
    fn is_destructive(&self) -> bool;
}

// 依赖注入注册
// ToolRegistryBuilder 声明 Resource:
// - Terminal: 终端后端
// - Cwd: 工作目录
// - ToolCallId: 调用 ID
// - SessionFolder: 输出路径
// - NotificationHandle: 通知通道
```

### 2.2 Smolagents — @tool 装饰器

```python
@tool
def web_search(query: str) -> str:
    """Search the web. Args: query: search query"""
    return search_api(query)

# 自动从 type hints + docstring 生成 JSON Schema
```

### 2.3 Cline SDK — createTool + Zod schema

```typescript
const tool = createTool({
    name: "deploy",
    description: "Deploy to staging",
    inputSchema: z.object({
        env: z.enum(["staging", "production"]),
        branch: z.string(),
    }),
    lifecycle: { completesRun: true },  // 调用后结束 Agent 运行
    execute: async (input) => { /* ... */ },
})
```

### 2.4 通用工具注册模式

```python
class ToolRegistry:
    """统一的工具注册中心"""
    def __init__(self):
        self._tools: dict[str, BaseTool] = {}

    def register(self, tool: BaseTool):
        self._tools[tool.name] = tool

    def to_openai_schema(self) -> list[dict]:
        """生成 OpenAI function calling JSON Schema"""
        return [t.to_schema() for t in self._tools.values()]

    def execute(self, name: str, args: dict) -> str:
        """按名称执行工具"""
        return self._tools[name].execute(**args)
```

---

## 三、记忆系统设计模式

### 3.1 Mem0 — 向量+图谱混合存储

来源: `mem0/memory/main.py`

```
┌─────────────────────────────────────────┐
│          add() 记忆添加流程               │
├─────────────────────────────────────────┤
│ Phase 0: 收集上下文（最近消息）           │
│ Phase 1: 向量搜索现有记忆 (top_k=10)     │
│ Phase 2: LLM 提取关键事实 → JSON         │
│ Phase 3: 批量向量化 (embed_batch)        │
│ Phase 4: MD5 哈希去重                    │
│ Phase 5: 向量存储批量插入               │
│ Phase 6: 实体提取 + 图谱链接             │
└─────────────────────────────────────────┘

┌─────────────────────────────────────────┐
│       search() 混合检索流程              │
├─────────────────────────────────────────┤
│ 1. Query 预处理（词形还原 + 实体提取）    │
│ 2. Query 向量化                         │
│ 3. 语义搜索 (over-fetch top_k*4)         │
│ 4. BM25 关键词搜索                       │
│ 5. 实体增强 (图谱 boosting)              │
│ 6. 混合评分: 语义 + BM25 + 实体 boost    │
│ 7. 可选重排序 (reranker)                 │
└─────────────────────────────────────────┘
```

**核心数据结构**:
```python
# 实体与记忆的多对多关系（图谱）
entity = {
    "data": entity_text,
    "entity_type": entity_type,
    "linked_memory_ids": [memory_id_1, memory_id_2, ...],
    "user_id": user_id,
}

# 记忆条目
memory = {
    "id": uuid,
    "memory": text,
    "hash": md5(text),       # 去重用
    "metadata": {...},
    "user_id": user_id,
}
```

### 3.2 OpenClaw — Markdown 文件式记忆

```
~/.openclaw/
├── SOUL.md       # Agent 人格定义
├── MEMORY.md     # 核心记忆（偏好、约定、知识）
├── HEARTBEAT.md  # 心跳任务检查
├── AGENTS.md     # Agent 配置
├── IDENTITY.md   # Agent 身份
├── USER.md       # 用户画像
├── memory/       # SQLite + BM25/向量混合检索
└── daily/        # 每日日志（append-only）
```

### 3.3 通用记忆系统设计

```python
class MemorySystem:
    """三层记忆架构"""
    def __init__(self):
        self.working = []           # 当前会话上下文
        self.semantic = VectorMemory()  # 向量语义记忆
        self.files = FileMemory()   # Markdown 文件记忆

    def remember(self, content: str, category: str):
        """写入三层记忆"""
        self.working.append(content)
        self.semantic.add(content, category)
        self.files.append(category, content)

    def recall(self, query: str) -> list[str]:
        """混合检索"""
        semantic_results = self.semantic.search(query)
        file_results = self.files.search(query)
        return merge_and_rank(semantic_results, file_results)

    def inject_context(self, query: str) -> str:
        """构建注入 LLM 的记忆上下文"""
        memories = self.recall(query)
        return format_as_prompt(memories)
```

---

## 四、LLM Provider 抽象层

### 4.1 Cline — @cline/llms 分层

```
@cline/llms (Provider 层)
  ├── ApiHandler 接口       ← 所有 Provider 实现此接口
  ├── ModelCatalog          ← 模型目录
  └── registerHandler()     ← 注册自定义 Provider

@cline/agents (无状态 Loop)
  └── 接收 ApiHandler → 运行 Agent Loop → 产生事件流

@cline/core (有状态 Runtime)
  └── 会话持久化 + Checkpoint + Cron + SubAgent
```

### 4.2 通用 Provider 抽象

```python
class LLMProvider(ABC):
    """统一 LLM 接口"""
    @abstractmethod
    def chat(self, messages: list, tools: list | None) -> LLMResponse:
        ...

    @abstractmethod
    def list_models(self) -> list[str]:
        ...

class OpenAIProvider(LLMProvider):
    base_url = "https://api.openai.com/v1"

class DeepSeekProvider(LLMProvider):
    base_url = "https://api.deepseek.com/v1"

class OllamaProvider(LLMProvider):
    base_url = "http://localhost:11434/v1"
```

---

## 五、SubAgent 子代理模式

### 5.1 Grok Build — Git Worktree 隔离

```rust
pub trait SubagentBackend {
    async fn spawn(&self, request: SubagentRequest) -> Result<SubagentResult>;
    async fn query(&self, id: &str, block: bool, timeout: Option<u64>) -> ...;
    async fn cancel(&self, target: SubagentCancelTarget) -> ...;
}

// 当前实现: ChannelBackend (tokio::mpsc)
// 未来: RemoteBackend（分布式子代理）
```

每个子代理创建独立 Git Worktree：环境隔离 + 状态持久化 + 资源调度。

### 5.2 Cline — Coordinator → Specialist

```typescript
// Coordinator agent breaks work into subtasks
// Each Specialist gets: own model, tools, context
// Team state persists across sessions
```

### 5.3 CrewAI — 角色-任务-团队

```python
researcher = Agent(role="研究员", goal="收集信息", tools=[search])
writer = Agent(role="写手", goal="撰写文章", tools=[write])

crew = Crew(agents=[researcher, writer], tasks=[...], process="sequential")
result = crew.kickoff()
```

---

## 六、安全与沙箱模式

### 6.1 Grok Build — 四级沙箱

| Profile | 文件读 | 文件写 | 网络 | 用例 |
|---------|--------|--------|------|------|
| `off` | 全部 | 全部 | 允许 | 无限制 |
| `workspace` | 全部 | CWD+/tmp/~/.grok/ | 允许 | 日常开发 |
| `read-only` | 全部 | 仅~/.grok/ | 阻止 | 代码审查 |
| `strict` | CWD+系统 | CWD+/tmp/~/.grok/ | 阻止 | 不信任代码 |

### 6.2 Smolagents — Executor 抽象

```python
class LocalPythonExecutor:  # 本地执行
class E2BExecutor:          # E2B 云端沙箱
class DockerExecutor:       # Docker 容器
class ModalExecutor:        # Modal 云端
class WASMExecutor:         # 浏览器 WebAssembly
```

### 6.3 Bash 命令安全过滤

```python
DANGEROUS = [
    r"rm\s+-rf\s+/",        # 递归删除根
    r"dd\s+if=",             # 磁盘操作
    r"mkfs\.",               # 格式化
    r">\s*/dev/sd",          # 写入块设备
    r"chmod\s+777\s+/",      # 权限全开
    r":(){ :|:& };:"         # Fork bomb
]
```

---

## 七、可观测性模式

### 7.1 Langfuse — OpenTelemetry 追踪

```
Agent Run
  ├── Turn 1
  │   ├── LLM Call (model, tokens, cost, latency)
  │   └── Tool Call (name, args, result, error)
  ├── Turn 2
  │   ├── LLM Call
  │   └── Tool Call
  └── Final Answer
```

### 7.2 追踪数据结构

```python
@dataclass
class TraceSpan:
    name: str
    trace_id: str
    start_time: float
    end_time: float
    events: list[dict]     # [{type: "llm_call", model, tokens, cost}, ...]
    children: list[TraceSpan]

class AgentTracer:
    def trace(self, name: str) -> TraceSpan:
        """创建追踪 Span"""
    def flush(self):
        """上报到 Langfuse"""
```

---

## 八、评估模式

### 8.1 Grok Build — 对抗性 Skeptic 验证

```
模型声明 update_goal(completed: true)
  ↓
Goal Tracker 创建验证任务
  ↓
Goal Classifier 启动 N 个 skeptic 子 Agent
  ↓
Majority Refute 投票
  ├── ≥2 skeptic 说"没完成" → 拒绝（生成 gap 摘要）
  └── ≥2 skeptic 说"完成了" → 接受
```

### 8.2 RAGAS — 标准化评估指标

```
答案相关性 (Answer Relevancy)
事实一致性 (Faithfulness)
检索精度 (Context Precision)
检索召回 (Context Recall)
```

---

## 九、MCP 协议实现

### 9.1 JSON-RPC 2.0 消息格式

```json
// Request
{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": "search", "arguments": {"q": "AI"}}}

// Response
{"jsonrpc": "2.0", "id": 1, "result": {"content": [{"type": "text", "text": "results..."}]}}

// Error
{"jsonrpc": "2.0", "id": 1, "error": {"code": -32000, "message": "Tool not found"}}
```

### 9.2 MCP 能力

| 能力 | 方法 | 用途 |
|------|------|------|
| Tools | `tools/list`, `tools/call` | 模型可调用的函数 |
| Resources | `resources/list`, `resources/read` | 模型可读取的数据 |
| Prompts | `prompts/list`, `prompts/get` | 预定义提示词模板 |

---

## 十、总结：做 Agent 的最优技术栈

| 层 | 推荐 | 理由 |
|----|------|------|
| Agent Loop | Smolagents 模式 | 1000行极简，ReAct + Code/Tool双Agent |
| 工具系统 | Grok Build trait + Smolagents @tool | 接口清晰 + 装饰器便捷 |
| 状态管理 | LangGraph StateGraph | Reducer + Checkpointer 生产级 |
| 记忆 | Mem0 + OpenClaw 文件 | 语义检索 + 人类可读 |
| LLM 接入 | LiteLLM | 100+模型统一 API |
| 子代理 | Grok Worktree + CrewAI 角色 | 隔离 + 角色分工 |
| 沙箱 | E2B + Docker | <200ms 冷启动 |
| 可观测性 | Langfuse OTEL | 全链路 Trace |
| 评估 | RAGAS + Grok Skeptic | 自动化 + 对抗验证 |
| 协议 | MCP (JSON-RPC 2.0) | 行业标准工具扩展 |