|
|
3 weeks ago | |
|---|---|---|
| .. | ||
| ref | 4 weeks ago | |
| README.md | 3 weeks ago | |
| decisions.md | 3 weeks ago | |
| multimodal.md | 1 month ago | |
| skills.md | 1 month ago | |
| tools.md | 4 weeks ago | |
| trace-api.md | 1 month ago | |
module/file.py:function_namedocs/decisions.md另行记录核心理念:所有 Agent 都是 Trace
| 类型 | 创建方式 | 父子关系 | 状态 |
|---|---|---|---|
| 主 Agent | 直接调用 runner.run() |
无 parent | 正常执行 |
| 子 Agent | 通过 subagent 工具 |
parent_trace_id / parent_goal_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 工具(计划管理)
│ ├── 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 # 技能加载
│ └── subagent.py # 子 Agent 统一入口(evaluate/delegate/explore)
│
├── 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/ | 执行追踪 + 计划管理 |
| tools/ | 与外部世界交互(文件、命令、网络、浏览器) |
| memory/ | 跨会话知识(Skills、Experiences) |
| llm/ | LLM Provider 适配 |
┌─────────────────────────────────────────────────────────────┐
│ Layer 3: Skills(技能库) │
│ - Markdown 文件,存储领域知识和能力描述 │
│ - 通过 skill 工具按需加载到对话历史 │
└─────────────────────────────────────────────────────────────┘
▲
│ 归纳
┌─────────────────────────────────────────────────────────────┐
│ Layer 2: Experience(经验库) │
│ - 数据库存储,条件 + 规则 + 证据 │
│ - 向量检索,注入到 system prompt │
└─────────────────────────────────────────────────────────────┘
▲
│ 提取
┌─────────────────────────────────────────────────────────────┐
│ Layer 1: Trace(任务状态) │
│ - 当前任务的工作记忆 │
│ - Trace + Messages 记录执行过程 │
│ - Goals 管理执行计划 │
└─────────────────────────────────────────────────────────────┘
Layer 1: Infrastructure(基础设施,AgentRunner 构造时设置)
trace_store, memory_store, tool_registry, llm_call, skills_dir, utility_llm_call
Layer 2: RunConfig(运行参数,每次 run 时指定)
├─ 模型层:model, temperature, max_iterations, tools
└─ 框架层:trace_id, agent_type, uid, system_prompt, parent_trace_id, ...
Layer 3: Messages(任务消息,OpenAI SDK 格式 List[Dict])
[{"role": "user", "content": "分析这张图的构图"}]
@dataclass
class RunConfig:
# 模型层参数
model: str = "gpt-4o"
temperature: float = 0.3
max_iterations: int = 200
tools: Optional[List[str]] = None # None = 全部内置工具
# 框架层参数
agent_type: str = "default"
uid: Optional[str] = None
system_prompt: Optional[str] = None # None = 从 skills 自动构建
enable_memory: bool = True
auto_execute_tools: bool = True
name: Optional[str] = None # 显示名称(空则由 utility_llm 自动生成)
# Trace 控制
trace_id: Optional[str] = None # None = 新建
parent_trace_id: Optional[str] = None # 子 Agent 专用
parent_goal_id: Optional[str] = None
# 续跑控制
insert_after: Optional[int] = None # 回溯插入点(message sequence)
实现:agent/core/runner.py:RunConfig
通过 RunConfig 参数自然区分,统一入口 run(messages, config):
| 模式 | trace_id | insert_after | messages 含义 |
|---|---|---|---|
| 新建 | None | - | 初始任务消息 |
| 续跑 | 已有 ID | None | 追加到末尾的新消息 |
| 回溯 | 已有 ID | 指定 sequence | 在插入点之后追加的新消息 |
async def run(messages: List[Dict], config: RunConfig = None) -> AsyncIterator[Union[Trace, Message]]:
# Phase 1: PREPARE TRACE
# 无 trace_id → 创建新 Trace(生成 name,初始化 GoalTree)
# 有 trace_id + 无 insert_after → 加载已有 Trace,状态置为 running
# 有 trace_id + 有 insert_after → 加载 Trace,执行 rewind(标记后续 msgs/goals 为 abandoned)
trace = await _prepare_trace(config)
yield trace
# Phase 2: BUILD HISTORY
# 加载已有 active messages(续跑/回溯场景)
# 构建 system prompt(新建时注入 skills/experiences;续跑时复用已有)
# 追加 input messages
history, sequence = await _build_history(trace, messages, config)
# Phase 3: AGENT LOOP
for iteration in range(config.max_iterations):
# 周期性注入 GoalTree + Active Collaborators(每 10 轮)
if iteration % 10 == 0:
inject_context(goal_tree, collaborators)
response = await llm_call(messages=history, model=config.model, tools=tool_schemas)
# 按需自动创建 root goal(兜底)
# 记录 assistant Message
# 执行工具,记录 tool Messages
# 无 tool_calls 则 break
# Phase 4: COMPLETE
# 更新 Trace 状态 (completed/failed)
trace.status = "completed"
yield trace
实现:agent/core/runner.py:AgentRunner
回溯通过 RunConfig(trace_id=..., insert_after=N) 触发,在 Phase 1 中执行:
status="abandoned"rewind 事件新消息的 sequence 从 max(all_sequences) + 1 开始,不复用被 abandon 的序号。
三种模式共享同一入口 run(messages, config):
# 新建
async for item in runner.run(
messages=[{"role": "user", "content": "分析项目架构"}],
config=RunConfig(model="gpt-4o"),
):
...
# 续跑:在已有 trace 末尾追加消息继续执行
async for item in runner.run(
messages=[{"role": "user", "content": "继续"}],
config=RunConfig(trace_id="existing-trace-id"),
):
...
# 回溯:从指定 sequence 处切断,插入新消息重新执行
# insert_after=5 表示保留 sequence ≤ 5 的消息,abandon 之后的,从此处开始
async for item in runner.run(
messages=[{"role": "user", "content": "换一个方案试试"}],
config=RunConfig(trace_id="existing-trace-id", insert_after=5),
):
...
insert_after 的值是 message 的 sequence 号,可通过 GET /api/traces/{trace_id}/messages 查看。如果指定的 sequence 是一条带 tool_calls 的 assistant 消息,系统会自动将截断点扩展到其所有对应的 tool response 之后(安全截断)。
run(messages, config):核心方法,流式返回 AsyncIterator[Union[Trace, Message]]run_result(messages, config):便利方法,内部消费 run(),返回结构化结果。主要用于 subagent 工具内部操作型端点(需在 api_server.py 中配置 Runner)。执行在后台异步进行,通过 WebSocket 监听进度。
| 方法 | 路径 | 说明 |
|---|---|---|
| POST | /api/traces |
新建 Trace 并执行 |
| POST | /api/traces/{id}/continue |
续跑 |
| POST | /api/traces/{id}/rewind |
回溯重放 |
| GET | /api/traces/running |
列出正在运行的 Trace |
# 新建
curl -X POST http://localhost:8000/api/traces \
-H "Content-Type: application/json" \
-d '{"messages": [{"role": "user", "content": "分析项目架构"}], "model": "gpt-4o"}'
# 续跑
curl -X POST http://localhost:8000/api/traces/{trace_id}/continue \
-d '{"messages": [{"role": "user", "content": "继续深入分析"}]}'
# 回溯:从 sequence 5 处截断,插入新消息重新执行
curl -X POST http://localhost:8000/api/traces/{trace_id}/rewind \
-d '{"insert_after": 5, "messages": [{"role": "user", "content": "换一个方案"}]}'
响应立即返回 {"trace_id": "...", "mode": "rewind", "status": "started"},通过 WS /api/traces/{trace_id}/watch 监听实时事件。
实现:agent/trace/run_api.py
一次完整的 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] = {} # 元数据(含 collaborators 列表)
# 当前焦点
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.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 | evaluate
sub_trace_metadata: Optional[Dict] = None
# 统计
self_stats: GoalStats # 自身 Messages 统计
cumulative_stats: GoalStats # 包含子孙的累计统计
created_at: datetime
Goal 类型:
normal - 普通目标,由 Agent 直接执行agent_call - 通过 subagent 工具创建的目标,会启动 Sub-Traceagent_call 类型的 Goal:
agent_call_mode 记录使用的模式(explore/delegate/evaluate)sub_trace_ids 记录创建的所有 Sub-Trace IDsummary 包含格式化的汇总结果(explore 模式会汇总所有分支)Goal 操作(通过 goal 工具):
add - 添加顶层目标under - 在指定目标下添加子目标after - 在指定目标后添加兄弟目标focus - 切换焦点到指定目标done - 完成当前目标(附带 summary)abandon - 放弃当前目标(附带原因)实现:agent/trace/goal_models.py, agent/trace/goal_tool.py
对应 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 # 全局顺序
status: Literal["active", "abandoned"] = "active" # 回溯时后续消息标记为 abandoned
goal_id: Optional[str] = None # 关联的 Goal ID(初始消息为 None,系统会按需自动创建 root goal 兜底)
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
abandoned_at: Optional[datetime] = None # 回溯标记时间
Message 提供格式转换方法:
to_llm_dict() → OpenAI 格式 Dict(用于 LLM 调用)from_llm_dict(d, trace_id, sequence, goal_id) → 从 OpenAI 格式创建 Message加载 messages 时,默认只返回 status="active" 的消息。
实现:agent/trace/models.py
不同类型 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 可覆盖或添加预设。
通过 subagent 工具创建子 Agent 执行任务,支持三种模式。子 Agent 通过 name 参数命名,便于后续引用和续跑。
subagent 工具负责创建 Sub-Trace 和初始化 GoalTree(因为需要设置自定义 context 元数据和命名规则),创建完成后将 trace_id 传给 RunConfig,由 Runner 接管后续执行。工具同时维护父 Trace 的 context["collaborators"] 列表。
并行探索多个分支,适合技术选型、方案对比等场景。
asyncio.gather() 并行执行所有分支委派单个任务给子 Agent 执行,适合代码分析、文档生成等场景。
name 续跑已有子 Agent:subagent(name="coder", task="继续")评估指定 Goal 的执行结果,提供质量评估和改进建议。
实现位置:agent/tools/builtin/subagent.py
详细文档:工具系统 - Subagent 工具
创建阻塞式 Trace,等待人类通过 IM/邮件等渠道回复。
注意:此功能规划中,暂未实现。
任务执行中与模型密切协作的实体(子 Agent 或人类),按 与当前任务的关系 分类,而非按 human/agent 分类:
| 持久存在(外部可查) | 任务内活跃(需要注入) | |
|---|---|---|
| Agent | 专用 Agent(代码审查等) | 当前任务创建的子 Agent |
| Human | 飞书通讯录 | 当前任务中正在对接的人 |
活跃协作者存储在 trace.context["collaborators"]:
{
"name": "researcher", # 名称(模型可见)
"type": "agent", # agent | human
"trace_id": "abc-@delegate-001", # trace_id(agent 场景)
"status": "completed", # running | waiting | completed | failed
"summary": "方案A最优", # 最近状态摘要
}
与 GoalTree 一同周期性注入(每 10 轮),渲染为 Markdown:
## Active Collaborators
- researcher [agent, completed]: 方案A最优
- 谭景玉 [human, waiting]: 已发送方案确认,等待回复
- coder [agent, running]: 正在实现特征提取模块
列表为空时不注入。
各工具负责更新 collaborators 列表(通过 context["store"] 写入 trace.context):
subagent 工具:创建/续跑子 Agent 时更新feishu 工具:发送消息/收到回复时更新持久联系人/Agent:通过工具按需查询(如 feishu_get_contact_list),不随任务注入。
实现:agent/core/runner.py:AgentRunner._build_context_injection, agent/tools/builtin/subagent.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 内部计划管理 |
builtin/ |
subagent | 子 Trace 创建(explore/delegate/evaluate) |
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
)
详细文档:工具系统
| 类型 | 加载位置 | 加载时机 |
|---|---|---|
| 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 使用指南
从执行历史中提取的经验规则,用于指导未来任务。
@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
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_trace_messages(self, trace_id: str, include_abandoned: bool = False) -> List[Message]: ...
async def get_messages_by_goal(self, trace_id: str, goal_id: str) -> List[Message]: ...
async def abandon_messages_after(self, trace_id: str, cutoff_sequence: int) -> List[str]: ...
async def append_event(self, trace_id: str, event_type: str, payload: Dict) -> int: ...
实现:
agent/trace/protocols.pyagent/trace/store.py:FileSystemTraceStore.trace/
├── {trace_id}/
│ ├── meta.json # Trace 元数据(含 tools 定义)
│ ├── goal.json # GoalTree(mission + goals 列表)
│ ├── events.jsonl # 事件流(goal 变更、sub_trace 生命周期等)
│ └── messages/ # Messages
│ ├── {trace_id}-0001.json
│ └── ...
│
└── {trace_id}@explore-{序号}-{timestamp}-001/ # 子 Trace
└── ...
events.jsonl 说明:
Sub-Trace 目录命名:
{parent}@explore-{序号:03d}-{timestamp}-001{parent}@delegate-{timestamp}-001{parent}@evaluate-{timestamp}-001meta.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": {
"collaborators": [
{"name": "researcher", "type": "agent", "trace_id": "...", "status": "completed", "summary": "方案A最优"}
]
},
"current_goal_id": "3"
}
详见 设计决策文档
核心决策:
所有 Agent 都是 Trace - 主 Agent、子 Agent、人类协助统一为 Trace,通过 parent_trace_id 和 spawn_tool 区分
trace/ 模块统一管理执行状态 - 合并原 execution/ 和 goal/,包含计划管理和 Agent 内部控制工具
tools/ 专注外部交互 - 文件、命令、网络、浏览器等与外部世界的交互
Agent 预设替代 Sub-Agent 配置 - 通过 core/presets.py 定义不同类型 Agent 的工具权限和参数
| 文档 | 内容 |
|---|---|
| Context 管理 | Goals、压缩、Plan 注入策略 |
| 工具系统 | 工具定义、注册、双层记忆 |
| Skills 指南 | Skill 分类、编写、加载 |
| 多模态支持 | 图片、PDF 处理 |
| 设计决策 | 架构决策记录 |
| 测试指南 | 测试策略和命令 |