可以。你列出的 10 个模块，基本已经覆盖了一个生产级 Agent 系统的核心架构。

但要真正理解它，不能只停留在模块职责层面，而应该回答三个问题：

1. 内部由什么组件组成？
2. 具体通过什么数据结构、算法和机制实现？
3. 各模块之间如何协作？

我建议把它作为一套 Agent Runtime 工程体系来学习和实现。

下面以 Python + FastAPI + Codex CLI / LLM API 为技术背景，讲清楚每个模块的具体实现机制。

一、整体架构

先把整体架构拆成四层。

Agent Runtime

任务入口 · 生命周期 · 状态机 · 调度器 · 执行引擎

Memory

记忆存储与检索

Planning

任务规划与拆解

Loop

推理与执行循环

Tool

工具注册与调用

Sandbox

隔离与安全执行

Context

上下文构建与预算

Model

模型路由与调用

Multi-Agent

子任务与协作

Human-in-the-loop

人工介入与审批

Observability

日志、Tracing、评估

基础设施层

PostgreSQL / Redis / Vector DB / Object Storage / Queue / Docker / Linux

这里有一个重要认识：

Agent Runtime 是调度和执行的核心，其他模块是它调用的能力组件。

例如：

• Runtime 决定什么时候执行任务。
• Planning 决定任务应该怎么做。
• Loop 决定下一步执行什么。
• Context 决定给模型提供什么信息。
• Tool 负责真正执行外部操作。

下面逐个拆解。

二、10 个核心模块的具体实现

1. Memory：记忆系统

核心目标：让 Agent 跨轮次、跨任务保留有价值的信息

Memory 不等于上下文。

上下文是本次模型调用所携带的信息，Memory 是可以跨模型调用、跨会话保存和检索的信息来源。

1.1 内部结构

建议分成四种记忆：

其中，Working Memory 和 Short-term Memory 不应该混为一谈：

• Working Memory 服务于当前任务执行。
• Short-term Memory 服务于当前会话理解。

1.2 具体实现机制

第一步：记忆写入

用户对话或者任务执行完成后，触发 Memory Extraction。

例如：

用户说：

我更喜欢简短的回答，不要展开太多。

记忆提取器判断这是一条长期有效的偏好。

写入数据库：

{
  "user_id": "user_001",
  "type": "preference",
  "content": "用户偏好简短、直接的回答",
  "importance": 0.9,
  "source": "conversation",
  "created_at": "2026-11-03"
}

第二步：记忆检索

新任务到来时，根据任务内容检索相关记忆。

通常采用混合检索：

用户当前请求
    |
    v
关键词检索（BM25）
    |
    +---- 向量检索（Embedding）
    |
    +---- 用户 / 时间 / 类型过滤
    |
    v
相关记忆排序
    |
    v
返回 Top-K

排序可以综合：

• 语义相似度
• 记忆重要程度
• 时间衰减
• 用户或项目关联性

第三步：记忆更新

不是每次都新增记录。

需要实现：

• 相同记忆合并
• 新信息覆盖旧信息
• 冲突记忆标记
• 过期记忆衰减
• 重要记忆保留

1.3 推荐技术

关键点：Memory 不应该直接把所有历史记录塞进 Prompt，而是检索出本次任务真正需要的内容。

2. Planning：任务规划系统

核心目标：把复杂任务拆解成可以执行的步骤

Planning 主要解决复杂任务怎么完成的问题。

例如用户提出：

帮我分析一个项目，研究竞品，最后生成一份 PPT。

Planning 不应该直接让模型一次性完成，而应该拆解：

任务：项目研究

├── Task 1：收集项目资料
├── Task 2：分析项目需求
├── Task 3：研究竞品
├── Task 4：整理分析结果
└── Task 5：生成 PPT

2.1 核心实现

Planning 本质上是：

LLM + 任务 DAG + 依赖管理 + 调度器。

首先通过 LLM 生成结构化任务计划：

{
  "goal": "完成项目研究报告",
  "tasks": [
    {
      "id": "task_1",
      "description": "收集项目资料",
      "dependencies": [],
      "status": "pending"
    },
    {
      "id": "task_2",
      "description": "研究竞品",
      "dependencies": ["task_1"],
      "status": "pending"
    },
    {
      "id": "task_3",
      "description": "生成研究报告",
      "dependencies": ["task_2"],
      "status": "pending"
    }
  ]
}

这里使用 DAG（有向无环图）管理任务依赖。

例如：

2.2 动态规划

真正的 Agent 不能只执行固定计划。

例如：

• Task 1 发现资料不足。
• Agent 判断需要补充搜索。
• Planning 增加一个新任务。
• 后续任务根据新的依赖关系调整。

因此需要支持：

class Planner:    async def create_plan(self, goal):        pass    async def update_plan(self, plan, observation):        pass    async def add_task(self, plan, task):        pass    async def cancel_task(self, task_id):        pass

注意：Planning 负责决定任务结构，不负责具体执行工具。

3. Loop：Agent 执行循环

核心目标：让 Agent 能够持续思考、调用工具、观察结果并决定下一步

这是 Agent 最核心的执行机制之一。

典型实现就是 ReAct Loop。

3.1 执行流程

3.2 具体实现

核心就是一个状态循环：

async def agent_loop(task):    while not task.is_terminal():        context = await context_manager.build(task)        response = await model.generate(context)        if response.type == "tool_call":            result = await tool_runtime.execute(                response.tool_call            )            await task_manager.update(                task.id,                result=result            )        elif response.type == "final":            await task_manager.complete(                task.id,                response.content            )        elif response.type == "human_approval":            await task_manager.pause(task.id)            return        elif response.type == "error":            await error_handler.handle(task, response)

生产环境还需要增加：

• 最大循环次数
• 最大执行时间
• Token 上限
• 工具调用次数限制
• 超时控制
• 重试策略
• 任务取消
• 中断恢复

3.3 状态机

建议单独实现 Task State Machine。

CREATED
   |
   v
PENDING
   |
   v
RUNNING
   |
   +----> WAITING_APPROVAL
   |              |
   |              v
   |           RUNNING
   |
   +----> WAITING_TOOL
   |              |
   |              v
   |           RUNNING
   |
   +----> COMPLETED
   |
   +----> FAILED
   |
   +----> CANCELLED

重要：Loop 是执行引擎，State Machine 是执行状态管理，两者应该分开。

4. Tool：工具系统

核心目标：让模型能够操作外部世界

Tool 本质上是一套工具注册、发现、权限控制和执行机制。

4.1 工具分类

4.2 工具注册机制

首先建立统一的 Tool Registry。

class ToolRegistry:    def __init__(self):        self.tools = {}    def register(self, tool):        self.tools[tool.name] = tool    def get(self, name):        return self.tools.get(name)    def list_tools(self):        return list(self.tools.values())

每个工具统一定义：

{    "name": "read_file",    "description": "读取文件",    "input_schema": {        "type": "object",        "properties": {            "path": {                "type": "string"            }        },        "required": ["path"]    }}

4.3 工具执行机制

LLM 返回 Tool Call
        |
        v
Tool Registry 查找工具
        |
        v
Permission Check
        |
        v
参数 Schema 校验
        |
        v
Tool Executor
        |
        v
执行工具
        |
        v
结果标准化
        |
        v
返回 Loop

需要注意几个工程问题：

1. 参数校验不能只依赖模型。
2. 工具执行必须有超时控制。
3. 危险操作需要权限校验。
4. 工具结果需要限制大小。
5. 工具调用要有审计日志。
6. 需要考虑幂等性，避免重试导致重复操作。

4.4 Codex 的工具 Runtime

你之前研究的 Codex 工具：

• exec_command
• write_stdin
• request_user_input
• view_image
• multi_agent_v1

可以理解为更完整的 Agent Tool Runtime。

特别是：

• exec_command：执行命令。
• write_stdin：与持续运行的进程交互。
• request_user_input：向用户请求输入。
• view_image：读取和理解图像。
• multi_agent_v1：协调子 Agent。

这里有一个区别：

Tool 是工具定义和调用机制，Tool Runtime 是工具的实际执行环境。

5. Sandbox：安全执行环境

核心目标：限制 Agent 对宿主机和外部资源的访问

如果 Agent 可以执行 Shell、修改文件、安装软件，就必须考虑隔离。

5.1 Sandbox 的层次

5.2 实际执行流程

例如用户要求：

帮我修改项目代码。

Agent 不能直接访问宿主机所有文件。

应该：

Agent Runtime
      |
      v
Sandbox Manager
      |
      v
创建任务隔离环境
      |
      v
挂载项目工作目录
      |
      v
限制文件系统访问
      |
      v
限制系统调用
      |
      v
限制 CPU / Memory
      |
      v
执行 Codex CLI
      |
      v
收集修改结果
      |
      v
销毁临时环境

5.3 具体技术选择

你目前采用的：

unshare
bwrap
seccomp

实际上是一个合理的 Linux 沙箱组合。

建议进一步抽象：

class SandboxManager:    async def create(self, task):        pass    async def execute(self, sandbox_id, command):        pass    async def collect_artifacts(self, sandbox_id):        pass    async def destroy(self, sandbox_id):        pass

还需要注意：Namespace 并不自动意味着安全隔离。必须正确配置 capabilities、UID 映射、挂载权限、网络策略和系统调用限制。

6. Context：上下文管理系统

核心目标：在有限的 Token 预算内，给模型提供最有价值的信息

这个模块非常重要。

它和 Memory 有明显区别：

• Memory：保存和检索信息。
• Context：组织本次模型调用需要的信息。

6.1 Context 由什么组成？

一次模型请求通常包含：

System Prompt
      +
Developer Instructions
      +
User Message
      +
Conversation History
      +
Memory Retrieval
      +
Task State
      +
Tool Definitions
      +
Tool Results
      +
Current Plan

这些内容组成最终发送给模型的上下文。

6.2 Context Builder

建议建立独立的 Context Builder。

class ContextBuilder:    async def build(self, task):        context = []        context += await self.get_system_prompt()        context += await self.get_task_state(task)        context += await self.get_memory(task)        context += await self.get_history(task)        context += await self.get_plan(task)        context += await self.get_tools(task)        return await self.optimize(context)

6.3 Token Budget

例如模型最大上下文窗口为 128K。

不能直接全部使用。

可以预留：

具体预算需要根据模型、任务和输出长度动态调整。

6.4 Context Compression

当上下文达到阈值时，需要压缩。

常见机制：

1. Sliding Window

保留最近 N 轮对话。

2. Summarization

将较早的对话压缩成摘要。

3. Tool Result Pruning

删除已经不再需要的大型工具结果。

4. Selective Retrieval

从 Memory 中重新检索需要的信息。

5. Hierarchical Summary

建立分层摘要：

完整历史记录
    |
    v
单轮摘要
    |
    v
任务摘要
    |
    v
会话摘要

这里特别注意：

Checkpoint 和 Context Compression 是两回事。

Checkpoint 保存的是执行恢复所需的状态快照，不代表一定要压缩上下文。

Checkpoint 可以保存完整的消息引用、工具调用状态、任务状态、执行位置等。

7. Model：模型管理系统

核心目标：根据任务需求选择合适的模型

不要在 Agent Loop 中直接写死某个模型 API。

应该抽象 Model Provider 和 Model Router。

7.1 模型管理结构

Model Router
     |
     +---- Task Classification
     |
     +---- Model Selection
     |
     +---- Fallback
     |
     +---- Retry
     |
     v
Model Provider
     |
     +---- OpenAI
     +---- DeepSeek
     +---- Claude
     +---- Qwen
     +---- Local Model

7.2 模型路由

例如：

路由依据可以是：

• 任务类型
• 复杂度
• Token 预算
• 模型能力
• 延迟要求
• 成本要求

7.3 Fallback

例如：

async def generate(messages):    try:        return await deepseek.generate(messages)    except TimeoutError:        return await openai.generate(messages)    except ModelUnavailable:        return await qwen.generate(messages)

实际生产环境还应该区分：

• 超时
• 限流
• 余额不足
• 服务不可用
• 参数错误
• 模型输出异常

不是所有异常都应该触发 Fallback。

8. Multi-Agent：多 Agent 协作系统

核心目标：将复杂任务分配给多个独立 Agent 执行

Multi-Agent 不等于同时运行多个 LLM。

核心是任务分配、独立执行、结果协调。

8.1 常见架构

我建议重点掌握三种。

模式一：Supervisor

主 Agent 负责分配任务，子 Agent 负责执行。

The Supervisor Pattern: Why Multi-Agent Systems Fail Without Coordination | by Naqeeb ali Shamsi | Feb, 2026 | Medium

naqeebali-shamsi.medium.com

Supervisor
    |
    +---- Research Agent
    |
    +---- Coding Agent
    |
    +---- Analysis Agent
    |
    v
Supervisor 汇总结果

模式二：Parallel

多个 Agent 并行执行独立任务。

          Main Agent
               |
       +-------+-------+
       |       |       |
      A1      A2      A3
       |       |       |
       +-------+-------+
               |
           Result

模式三：Pipeline

多个 Agent 按照流程执行。

Research Agent
      |
      v
Analysis Agent
      |
      v
Writing Agent
      |
      v
Review Agent

8.2 具体实现机制

需要实现：

1. Agent Registry：管理 Agent。
2. Agent Profile：定义角色、模型和工具。
3. Task Dispatcher：分配子任务。
4. Execution Manager：管理并发执行。
5. Message Bus：Agent 间通信。
6. Result Aggregator：收集结果。
7. Shared State：共享必要的任务数据。

例如：

async def run_parallel(tasks):    results = await asyncio.gather(        *[            agent.execute(task)            for task in tasks        ],        return_exceptions=True    )    return results

实际生产中还需要并发限制、失败隔离、取消传播、子任务超时和结果校验。

我的建议：先实现 Supervisor + Parallel，再考虑复杂的 Agent 自主协商。

9. Human-in-the-loop：人工介入系统

核心目标：让人类能够在 Agent 执行过程中参与决策

Agent 不应该完全自主执行所有操作。

例如：

• 删除文件。
• 执行高风险命令。
• 发布代码。
• 发送邮件。
• 产生付费行为。

这些操作可以要求用户确认。

9.1 实现机制

核心是 Interrupt + Resume。

Agent 执行任务
      |
      v
遇到高风险操作
      |
      v
创建 Approval Request
      |
      v
保存任务 Checkpoint
      |
      v
WAITING_APPROVAL
      |
      v
用户确认
      |
      v
恢复任务
      |
      v
继续执行

9.2 数据结构

{
  "approval_id": "approval_001",
  "task_id": "task_001",
  "action": "delete_file",
  "arguments": {
    "path": "/workspace/test.py"
  },
  "status": "pending",
  "created_at": "2026-11-03"
}

用户批准后：

async def resume_task(task_id, approval):    if approval.status == "approved":        await task_manager.resume(task_id)    else:        await task_manager.reject(task_id)

9.3 必须区分三种状态

• Pause：主动暂停。
• Interrupt：执行过程被中断。
• Resume：从保存的状态恢复。

另外，审批恢复时需要重新校验用户权限和待执行操作，不能直接信任之前缓存的授权结果。

10. Observability：可观测性系统

核心目标：知道 Agent 为什么这样执行，以及执行得怎么样

传统 Web 服务主要监控：

• 请求量
• 响应时间
• 错误率
• CPU / Memory

但 Agent 还需要监控模型和工具执行过程。

10.1 核心监控指标

10.2 Tracing

建议采用 OpenTelemetry。

一个任务的 Trace：

Task: task_001
  |
  +-- Planning
  |
  +-- Context Build
  |
  +-- LLM Call #1
  |
  +-- Tool: web_search
  |
  +-- LLM Call #2
  |
  +-- Tool: read_file
  |
  +-- LLM Call #3
  |
  +-- Final Answer

每个步骤都应该记录：

{
  "task_id": "task_001",
  "span_id": "span_003",
  "operation": "llm_call",
  "model": "deepseek-v4",
  "input_tokens": 1200,
  "output_tokens": 300,
  "latency_ms": 1200,
  "status": "success"
}

10.3 Evaluation

Observability 不只是监控运行情况，还应该评估 Agent 的实际能力。

建议增加：

• 任务完成率
• 工具选择准确率
• 工具参数准确率
• 结果正确率
• 幻觉率
• 人工干预率
• 单任务成本

这样才能形成持续优化闭环。

三、10 个模块之间究竟如何协作？

上面介绍的是独立模块，现在把它们串起来。

假设用户给 Agent 一个任务：

分析某个 GitHub 项目，修改代码并生成报告。

完整执行流程：

这里有两个横向贯穿所有模块的能力：

1. 状态管理：任务执行到哪里、哪些工具已经执行、哪些任务还没完成。
2. Observability：每个模块的执行过程都应该留下可追踪的数据。

四、真正落地时，应该设计哪些核心类？

我建议不要直接按照 10 个模块创建 10 个 Python 文件。

应该先设计核心接口。

建议使用以下工程结构：

agent-runtime/
│
├── runtime/
│   ├── engine.py
│   ├── task_manager.py
│   ├── state_machine.py
│   └── scheduler.py
│
├── memory/
│   ├── manager.py
│   ├── retriever.py
│   ├── extractor.py
│   └── storage.py
│
├── planning/
│   ├── planner.py
│   ├── task_graph.py
│   └── scheduler.py
│
├── loop/
│   ├── agent_loop.py
│   ├── react.py
│   └── error_handler.py
│
├── tools/
│   ├── registry.py
│   ├── executor.py
│   ├── mcp_client.py
│   └── skill_loader.py
│
├── sandbox/
│   ├── manager.py
│   ├── namespace.py
│   ├── seccomp.py
│   └── resource_limits.py
│
├── context/
│   ├── builder.py
│   ├── compressor.py
│   ├── token_budget.py
│   └── history.py
│
├── models/
│   ├── router.py
│   ├── provider.py
│   └── fallback.py
│
├── multi_agent/
│   ├── manager.py
│   ├── dispatcher.py
│   └── aggregator.py
│
├── human/
│   ├── approval.py
│   └── interrupt.py
│
├── observability/
│   ├── tracing.py
│   ├── metrics.py
│   └── evaluation.py
│
└── infrastructure/
    ├── database.py
    ├── redis.py
    ├── queue.py
    └── storage.py