Agent 核心框架:提供单个 Agent 的执行能力
module/file.py:function_namedocs/decisions.md另行记录Agent Core 是一个完整的 Agent 执行框架,提供:
独立性:Agent Core 不依赖任何其他模块,可以独立运行。
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
│
├── tools/ # 外部交互工具
│ ├── registry.py # 工具注册表
│ ├── schema.py # Schema 生成器
│ ├── models.py # ToolResult, ToolContext
│ └── builtin/
│ ├── file/ # 文件操作
│ ├── browser/ # 浏览器自动化
│ ├── bash.py # 命令执行
│ ├── subagent.py # 子 Agent 创建
│ └── a2a_im.py # A2A IM 工具(桥接到 Gateway)
│
├── skill/ # 技能系统
│ ├── models.py # Skill
│ ├── skill_loader.py # Skill 加载器
│ └── skills/ # 内置 Skills
│
└── llm/ # LLM 集成
├── gemini.py # Gemini Provider
├── openrouter.py # OpenRouter Provider
└── yescode.py # Yescode Provider
一次完整的 Agent 执行。所有 Agent(主、子、人类协助)都是 Trace。
实现位置:agent/trace/models.py:Trace
计划中的一个目标,支持层级结构。
实现位置:agent/trace/goal_models.py:Goal
对应 LLM API 的消息,每条 Message 关联一个 Goal。
实现位置:agent/trace/models.py:Message
from agent.core.runner import AgentRunner, RunConfig
from agent.trace import FileSystemTraceStore, Trace, Message
from agent.llm import create_qwen_llm_call
runner = AgentRunner(
llm_call=create_qwen_llm_call(model="qwen3.5-plus"),
trace_store=FileSystemTraceStore(base_path=".trace"),
skills_dir="./skills", # 项目 skills 目录(可选)
)
async for item in runner.run(
messages=[{"role": "user", "content": "分析项目架构"}],
config=RunConfig(model="qwen3.5-plus"),
):
if isinstance(item, Trace):
print(f"Trace: {item.trace_id}")
elif isinstance(item, Message) and item.role == "assistant":
print(item.content)
RunConfig(
model="qwen3.5-plus",
temperature=0.3,
max_iterations=200,
agent_type="default", # Agent 预设(对应 presets.json)
name="任务名称",
tools=["read_file", "bash"], # 限制可用工具(None=全部)
goal_compression="on_overflow", # Goal 压缩:"none" / "on_complete" / "on_overflow"
knowledge=KnowledgeConfig(...), # 知识提取配置
)
config = RunConfig(model="qwen3.5-plus", trace_id="existing-trace-id")
async for item in runner.run(messages=[{"role": "user", "content": "继续"}], config=config):
...
同一个长期续跑的 agent,不同调用可以注入不同 skill:
# 不同调用场景注入不同策略
async for item in runner.run(
messages=[...], config=config,
inject_skills=["ask_strategy"], # 本次需要的 skill
skill_recency_threshold=10, # 最近 N 条消息内有就不重复注入
):
...
from agent.tools import tool, ToolContext, ToolResult
@tool(description="自定义工具")
async def my_tool(arg: str, ctx: ToolContext) -> ToolResult:
return ToolResult(title="成功", output=f"处理结果: {arg}")
参考 examples/research/ — 一个完整的调研 agent 项目:
examples/research/
├── run.py # 入口(含交互控制、续跑、暂停)
├── config.py # RunConfig + KnowledgeConfig 配置
├── presets.json # Agent 预设(工具权限、skills 过滤)
├── requirement.prompt # 任务 prompt($system$ + $user$ 段)
├── skills/ # 项目自定义 skills
└── tools/ # 项目自定义工具
| 文档 | 内容 |
|---|---|
| 架构设计 | 框架完整架构 |
| Context 管理 | 注入机制、压缩策略、Skill 指定注入 |
| 工具系统 | 工具定义、注册、参数注入 |
| Skills 指南 | Skill 分类、编写、加载 |
| Prompt 规范 | Prompt 撰写原则 |
| Trace API | REST API 和 WebSocket 接口 |
| 设计决策 | 架构决策记录 |
| 方法 | 路径 | 说明 |
|---|---|---|
| GET | /api/traces |
列出 Traces |
| GET | /api/traces/{id} |
获取 Trace 详情 |
| GET | /api/traces/{id}/messages |
获取 Messages |
| POST | /api/traces |
新建 Trace 并执行 |
| POST | /api/traces/{id}/run |
续跑或回溯 |
| POST | /api/traces/{id}/stop |
停止运行 |
实现:agent/trace/api.py, agent/trace/run_api.py