Agent/docs

Agent 功能需求与架构设计文档

文档维护规范

1. 先改文档，再动代码 - 新功能或重大修改需先完成文档更新、并完成审阅后，再进行代码实现；除非改动较小、不被文档涵盖
2. 文档分层，链接代码 - 重要或复杂设计可以另有详细文档；关键实现需标注代码文件路径；格式：module/file.py:function_name
3. 简洁快照，日志分离 - 只记录最重要的、与代码准确对应的或者明确的已完成的设计的信息，避免推测、建议，或大量代码；决策依据或修改日志若有必要，可在docs/decisions.md另行记录

---

系统概览

核心理念：所有 Agent 都是 Trace

类型	创建方式	父子关系	状态
主 Agent	直接调用 runner.run()	无 parent	正常执行
子 Agent	通过 task 工具	parent_trace_id 指向父	正常执行
人类协助	通过 ask_human 工具	parent_trace_id 指向父	阻塞等待

---

核心架构

模块结构

agent/
├── core/                  # 核心引擎
│   ├── runner.py          # AgentRunner + 运行时配置
│   └── presets.py         # Agent 预设（explore、analyst 等）
│
├── trace/                 # 执行追踪（含计划管理）
│   ├── models.py          # Trace, Message
│   ├── goal_models.py     # Goal, GoalTree, GoalStats
│   ├── protocols.py       # TraceStore 接口
│   ├── store.py           # FileSystemTraceStore 实现
│   ├── goal_tool.py       # goal 工具（计划管理）
│   ├── explore.py         # 探索式子任务
│   ├── delegate.py        # 委派式子任务
│   ├── compaction.py      # Context 压缩
│   ├── api.py             # REST API
│   ├── websocket.py       # WebSocket API
│   └── trace_id.py        # Trace ID 生成工具
│
├── tools/                 # 外部交互工具
│   ├── registry.py        # 工具注册表
│   ├── schema.py          # Schema 生成器
│   ├── models.py          # ToolResult, ToolContext
│   └── builtin/
│       ├── file/          # 文件操作（read, write, edit, glob, grep）
│       ├── browser/       # 浏览器自动化
│       ├── bash.py        # 命令执行
│       ├── sandbox.py     # 沙箱环境
│       ├── search.py      # 网络搜索
│       ├── webfetch.py    # 网页抓取
│       ├── skill.py       # 技能加载
│       └── ask_human.py   # 人类协助
│
├── memory/                # 跨会话记忆
│   ├── models.py          # Experience, Skill
│   ├── protocols.py       # MemoryStore 接口
│   ├── stores.py          # 存储实现
│   ├── skill_loader.py    # Skill 加载器
│   └── skills/            # 内置 Skills
│       └── core.md        # Core Skill（自动加载）
│
├── llm/                   # LLM 集成
│   ├── gemini.py          # Gemini Provider
│   ├── openrouter.py      # OpenRouter Provider
│   └── prompts/           # Prompt 工具

职责划分

模块	职责
core/	Agent 执行引擎 + 预设配置
trace/	执行追踪 + 计划管理 + 子任务（goal, explore, delegate）
tools/	与外部世界交互（文件、命令、网络、浏览器）
memory/	跨会话知识（Skills、Experiences）
llm/	LLM Provider 适配

三层记忆模型

┌─────────────────────────────────────────────────────────────┐
│ Layer 3: Skills（技能库）                                     │
│ - Markdown 文件，存储领域知识和能力描述                        │
│ - 通过 skill 工具按需加载到对话历史                            │
└─────────────────────────────────────────────────────────────┘
                              ▲
                              │ 归纳
┌─────────────────────────────────────────────────────────────┐
│ Layer 2: Experience（经验库）                                 │
│ - 数据库存储，条件 + 规则 + 证据                              │
│ - 向量检索，注入到 system prompt                              │
└─────────────────────────────────────────────────────────────┘
                              ▲
                              │ 提取
┌─────────────────────────────────────────────────────────────┐
│ Layer 1: Trace（任务状态）                                    │
│ - 当前任务的工作记忆                                          │
│ - Trace + Messages 记录执行过程                               │
│ - Goals 管理执行计划                                          │
└─────────────────────────────────────────────────────────────┘

---

核心流程：Agent Loop

async def run(task: str, agent_type: str = "default") -> AsyncIterator[Union[Trace, Message]]:
    # 1. 创建 Trace
    trace = Trace.create(
        mode="agent",
        task=task,
        agent_type=agent_type,
        model=config.model
    )
    await store.create_trace(trace)
    yield trace

    # 2. 加载 Skills，构建 system prompt
    skills = load_skills_from_dir(skills_dir)
    system_prompt = build_system_prompt(skills)

    # 3. 初始化
    messages = [{"role": "user", "content": task}]

    # 4. ReAct 循环
    for step in range(max_iterations):
        # 注入当前计划
        plan_text = goal_tree.to_prompt()

        # 调用 LLM
        response = await llm.chat(
            messages=messages,
            system=system_prompt + plan_text,
            tools=tool_registry.to_schema()
        )

        # 记录 assistant Message
        await store.add_message(Message.create(
            trace_id=trace.trace_id,
            role="assistant",
            sequence=next_seq,
            content=response
        ))
        yield assistant_msg

        # 没有工具调用，完成
        if not response.tool_calls:
            break

        # 执行工具，记录 tool Message
        for tool_call in response.tool_calls:
            result = await execute_tool(tool_call)
            await store.add_message(Message.create(
                trace_id=trace.trace_id,
                role="tool",
                sequence=next_seq,
                content=result
            ))
            yield tool_msg

    # 5. 完成
    trace.status = "completed"
    yield trace

实现：agent/core/runner.py:AgentRunner

---

数据模型

Trace（任务执行）

一次完整的 Agent 执行。所有 Agent（主、子、人类协助）都是 Trace。

@dataclass
class Trace:
    trace_id: str
    mode: Literal["call", "agent"]           # 单次调用 or Agent 模式

    # Prompt 标识
    prompt_name: Optional[str] = None

    # Agent 模式特有
    task: Optional[str] = None
    agent_type: Optional[str] = None

    # 父子关系（Sub-Trace 特有）
    parent_trace_id: Optional[str] = None    # 父 Trace ID
    parent_goal_id: Optional[str] = None     # 哪个 Goal 启动的

    # 状态
    status: Literal["running", "completed", "failed"] = "running"

    # 统计
    total_messages: int = 0
    total_tokens: int = 0                    # 总 tokens（prompt + completion）
    total_prompt_tokens: int = 0
    total_completion_tokens: int = 0
    total_cost: float = 0.0
    total_duration_ms: int = 0

    # 进度追踪
    last_sequence: int = 0                   # 最新 message 的 sequence
    last_event_id: int = 0                   # 最新事件 ID（用于 WS 续传）

    # 配置
    uid: Optional[str] = None
    model: Optional[str] = None              # 默认模型
    tools: Optional[List[Dict]] = None       # 工具定义（OpenAI 格式）
    llm_params: Dict[str, Any] = {}          # LLM 参数（temperature 等）
    context: Dict[str, Any] = {}             # 其他元数据

    # 当前焦点
    current_goal_id: Optional[str] = None

    # 结果
    result_summary: Optional[str] = None
    error_message: Optional[str] = None

    # 时间
    created_at: datetime
    completed_at: Optional[datetime] = None

实现：agent/trace/models.py

Goal（目标节点）

计划中的一个目标，支持层级结构。单独存储于 goal.json。

@dataclass
class Goal:
    id: str                                  # 内部 ID（"1", "2"...）
    description: str
    reason: str = ""                         # 创建理由
    parent_id: Optional[str] = None          # 父 Goal ID
    type: GoalType = "normal"                # normal | agent_call
    status: GoalStatus = "pending"           # pending | in_progress | completed | abandoned
    summary: Optional[str] = None            # 完成/放弃时的总结

    # agent_call 特有（启动 Sub-Trace）
    sub_trace_ids: Optional[List[str]] = None
    agent_call_mode: Optional[str] = None    # explore | delegate | sequential
    sub_trace_metadata: Optional[Dict] = None

    # 统计
    self_stats: GoalStats                    # 自身 Messages 统计
    cumulative_stats: GoalStats              # 包含子孙的累计统计

    created_at: datetime

Goal 操作（通过 goal 工具）：

• add - 添加顶层目标
• under - 在指定目标下添加子目标
• after - 在指定目标后添加兄弟目标
• focus - 切换焦点到指定目标
• done - 完成当前目标（附带 summary）
• abandon - 放弃当前目标（附带原因）

实现：agent/trace/goal_models.py, agent/trace/goal_tool.py

Message（执行消息）

对应 LLM API 的消息，每条 Message 关联一个 Goal。

@dataclass
class Message:
    message_id: str                          # 格式：{trace_id}-{sequence:04d}
    trace_id: str
    role: Literal["system", "user", "assistant", "tool"]
    sequence: int                            # 全局顺序
    goal_id: Optional[str] = None            # 关联的 Goal ID
    description: str = ""                    # 系统自动生成的摘要
    tool_call_id: Optional[str] = None
    content: Any = None

    # 统计
    prompt_tokens: Optional[int] = None
    completion_tokens: Optional[int] = None
    cost: Optional[float] = None
    duration_ms: Optional[int] = None

    # LLM 响应信息（仅 role="assistant"）
    finish_reason: Optional[str] = None

    created_at: datetime

实现：agent/trace/models.py

---

Agent 预设

不同类型 Agent 的配置模板，控制工具权限和参数。

@dataclass
class AgentPreset:
    allowed_tools: Optional[List[str]] = None  # None 表示允许全部
    denied_tools: Optional[List[str]] = None   # 黑名单
    max_iterations: int = 30
    temperature: Optional[float] = None
    description: Optional[str] = None


AGENT_PRESETS = {
    "default": AgentPreset(
        allowed_tools=None,
        max_iterations=30,
        description="默认 Agent，拥有全部工具权限",
    ),
    "explore": AgentPreset(
        allowed_tools=["read", "glob", "grep", "list_files"],
        denied_tools=["write", "edit", "bash", "task"],
        max_iterations=15,
        description="探索型 Agent，只读权限，用于代码分析",
    ),
    "analyst": AgentPreset(
        allowed_tools=["read", "glob", "grep", "web_search", "webfetch"],
        denied_tools=["write", "edit", "bash", "task"],
        temperature=0.3,
        max_iterations=25,
        description="分析型 Agent，用于深度分析和研究",
    ),
}

实现：agent/core/presets.py

用户自定义：项目级配置 .agent/presets.json 可覆盖或添加预设。

---

子 Trace 机制

explore 模式

并行探索多个分支，汇总结果返回。

# 创建多个探索分支
child_traces = await create_explore_traces(
    parent_trace_id=ctx.trace_id,
    prompts=["分析 A 方案", "分析 B 方案", "分析 C 方案"]
)
# 并行执行
results = await asyncio.gather(*[run_trace(t) for t in child_traces])

实现：agent/trace/explore.py

delegate 模式

委派子任务给专门的 Agent 执行。

child_trace = await create_delegate_trace(
    parent_trace_id=ctx.trace_id,
    task=prompt,
    agent_type="analyst"  # 使用 analyst 预设
)
result = await run_trace(child_trace)

实现：agent/trace/delegate.py

ask_human 工具

创建阻塞式 Trace，等待人类通过 IM/邮件等渠道回复。

@tool(name="ask_human")
async def ask_human_tool(question: str, channel: str, ctx: ToolContext) -> ToolResult:
    human_trace = await ctx.store.create_trace(
        task=question,
        parent_trace_id=ctx.trace_id,
        blocked=True,
        blocked_reason=f"等待人类通过 {channel} 回复"
    )
    await send_to_channel(channel, question)
    response = await wait_for_human_response(human_trace.id)
    return ToolResult(output=response)

实现：agent/tools/builtin/ask_human.py

---

工具系统

核心概念

@tool()
async def my_tool(arg: str, ctx: ToolContext) -> ToolResult:
    return ToolResult(
        title="Success",
        output="Result content",
        long_term_memory="Short summary"  # 可选：压缩后保留的摘要
    )

类型	作用
@tool	装饰器，自动注册工具并生成 Schema
ToolResult	工具执行结果，支持双层记忆
ToolContext	工具执行上下文，依赖注入

工具分类

目录	工具	说明
trace/	goal	Agent 内部计划管理
trace/	explore, delegate	子 Trace 创建
builtin/file/	read, write, edit, glob, grep	文件操作
builtin/browser/	browser actions	浏览器自动化
builtin/	bash, sandbox, search, webfetch, skill, ask_human	其他工具

双层记忆管理

大输出（如网页抓取）只传给 LLM 一次，之后用摘要替代：

ToolResult(
    output="<10K tokens 的完整内容>",
    long_term_memory="Extracted 10000 chars from amazon.com",
    include_output_only_once=True
)

详细文档：工具系统

---

Skills 系统

分类

类型	加载位置	加载时机
Core Skill	System Prompt	Agent 启动时自动加载
普通 Skill	对话消息	模型调用 skill 工具时

目录结构

agent/memory/skills/
├── core.md              # Core Skill（自动加载到 System Prompt）
└── browser_use/         # 普通 Skill（按需加载）

./skills/                # 项目自定义 Skills（按需加载）

实现：agent/memory/skill_loader.py

详细文档：Skills 使用指南

---

Experiences 系统

从执行历史中提取的经验规则，用于指导未来任务。

数据结构

@dataclass
class Experience:
    id: str
    scope: str           # "agent:executor" 或 "user:123"
    condition: str       # "当遇到数据库连接超时"
    rule: str            # "增加重试次数到5次"
    evidence: Dict       # 证据（trace_ids）
    confidence: float
    usage_count: int
    success_rate: float
    embedding: List[float]  # 向量，用于检索

检索和注入

# 1. 检索相关 Experiences
experiences = await db.query(
    "SELECT * FROM experiences WHERE scope = $1 ORDER BY embedding <-> $2 LIMIT 10",
    f"agent:{agent_type}", embed(task)
)

# 2. 注入到 system prompt
system_prompt += "\n# Learned Experiences\n" + format_experiences(experiences)

存储：PostgreSQL + pgvector

实现：agent/memory/stores.py:ExperienceStore

---

Context 压缩

压缩时机

Goal 完成（done）或放弃（abandon）时，将详细 Messages 替换为 Summary Message。

压缩策略

Goal 状态变化
    ↓
收集该 Goal 下的所有 Messages
    ↓
生成 Summary（由 LLM 提供）
    ↓
替换原始 Messages 为单条 Summary Message
    ↓
更新统计信息

实现：agent/trace/compaction.py

详细文档：Context 管理

---

存储接口

class TraceStore(Protocol):
    async def create_trace(self, trace: Trace) -> None: ...
    async def get_trace(self, trace_id: str) -> Trace: ...
    async def update_trace(self, trace_id: str, **updates) -> None: ...
    async def add_message(self, message: Message) -> None: ...
    async def get_messages(self, trace_id: str) -> List[Message]: ...
    async def get_messages_by_goal(self, trace_id: str, goal_id: str) -> List[Message]: ...

实现：

• 协议定义：agent/trace/protocols.py
• 文件存储：agent/trace/store.py:FileSystemTraceStore

存储结构

.trace/
├── {trace_id}/
│   ├── meta.json        # Trace 元数据（含 tools 定义）
│   ├── goal.json        # GoalTree（mission + goals 列表）
│   └── messages/        # Messages
│       ├── 0001.json
│       └── ...
│
└── {trace_id}@explore-{timestamp}-001/  # 子 Trace
    └── ...

meta.json 示例：

{
  "trace_id": "0415dc38-...",
  "mode": "agent",
  "task": "分析代码结构",
  "agent_type": "default",
  "status": "running",
  "model": "google/gemini-2.5-flash",
  "tools": [...],
  "llm_params": {"temperature": 0.3},
  "context": {},
  "current_goal_id": "3"
}

---

设计决策

详见 设计决策文档

核心决策：

1. 所有 Agent 都是 Trace - 主 Agent、子 Agent、人类协助统一为 Trace，通过 parent_trace_id 和 spawn_tool 区分

2. trace/ 模块统一管理执行状态 - 合并原 execution/ 和 goal/，包含计划管理和 Agent 内部控制工具

3. tools/ 专注外部交互 - 文件、命令、网络、浏览器等与外部世界的交互

4. Agent 预设替代 Sub-Agent 配置 - 通过 core/presets.py 定义不同类型 Agent 的工具权限和参数

---

相关文档

文档	内容
Context 管理	Goals、压缩、Plan 注入策略
工具系统	工具定义、注册、双层记忆
Skills 指南	Skill 分类、编写、加载
多模态支持	图片、PDF 处理
设计决策	架构决策记录
测试指南	测试策略和命令