Agent/cyber_agent

Agent Core

Agent 核心框架：提供单个 Agent 的执行能力

文档维护规范

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

---

概述

Agent Core 是一个完整的 Agent 执行框架，提供：

• Trace、Message、Goal 管理
• 工具系统（文件、命令、网络、浏览器）
• LLM 集成（Qwen、Gemini、Claude、OpenRouter、Yescode）
• Skills（领域知识注入）
• 子 Agent 机制

独立性：Agent Core 不依赖任何其他模块，可以独立运行。

---

模块结构

cyber_agent/
├── core/                  # 核心引擎
│   ├── runner.py          # AgentRunner + 运行时配置
│   ├── agent_mode.py      # Legacy / Recursive 模式策略
│   ├── task_protocol.py   # Recursive 任务报告与审核协议
│   └── presets.py         # Agent 预设（delegate、explore、evaluate 等）
│
├── 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 集成
    ├── qwen.py           # Qwen / 百炼 Provider
    ├── gemini.py          # Gemini Provider
    ├── claude.py          # Anthropic 原生 Provider
    ├── openrouter.py      # OpenRouter Provider
    └── yescode.py         # Yescode Provider

---

核心概念

Trace（任务执行）

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

实现位置：cyber_agent/trace/models.py:Trace

Goal（目标节点）

计划中的一个目标，支持层级结构。

实现位置：cyber_agent/trace/goal_models.py:Goal

Message（执行消息）

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

实现位置：cyber_agent/trace/models.py:Message

---

应用框架设计（M1）

M1 只建立业务应用接入的设计基线，尚未发布可导入的 Application API，也没有可运行的 reference example：

• 应用框架能力、边界与状态权威基线（框架团队内部）
• Application API 0.1 草案（业务团队可读，未实现）
• Application Reference Example 验收合同（后续 M2/M3 实现输入）
• Decision 26：应用框架边界、状态权威与版本化装配

当前业务接入仍使用下方的 AgentRunner / RunConfig 接口；不得从 M1 草案中 import 尚未实现的类型。

---

快速开始

安装与环境变量

python3.11 -m venv .venv
source .venv/bin/activate
pip install -e . openai
cp .env.template .env

Qwen 最小配置：

QWEN_API_KEY=...
QWEN_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
AGENT_MODE=legacy

代码读取的是 QWEN_API_KEY；如果已有百炼 BAILIAN_API_KEY，需将其值同步给 QWEN_API_KEY。AGENT_MODE=recursive 还要在 RunConfig.root_task_anchor 中提供根任务目标、完成标准和硬约束。

基础使用

import asyncio
from dotenv import load_dotenv

from cyber_agent.core.runner import AgentRunner, RunConfig
from cyber_agent.trace import FileSystemTraceStore, Trace, Message
from cyber_agent.llm import create_qwen_llm_call

load_dotenv()

async def main():
    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)

asyncio.run(main())

Recursive 最小示例

先在环境中选择模式：

AGENT_MODE=recursive

新建 Recursive 根 Trace 必须显式提供根任务锚点。在上例的 main() 中将 runner.run(...) 替换为：

async for item in runner.run(
    messages=[{"role": "user", "content": "分析项目架构并给出依赖关系"}],
    config=RunConfig(
        model="qwen3.5-plus",
        root_task_anchor={
            "objective": "分析项目架构并给出依赖关系",
            "completion_criteria": [
                "列出主要模块及职责",
                "说明核心调用链和模块依赖",
            ],
            "constraints": ["结论必须来自实际代码"],
        },
    ),
):
    ...

Recursive 子任务使用结构化 TaskBrief，并经报告、独立验收和父级审核后继续规划。根锚点与授权引用见 Context 管理，深度、工具权限、调度、停止和资源预算见 Agent 运行模式，报告与审核见 Recursive 任务协议。

其他 Provider 通过对应的 create_*_llm_call() 创建，再传入 AgentRunner(llm_call=...)：

Provider	创建函数	环境变量
Qwen	create_qwen_llm_call()	QWEN_API_KEY，可选 QWEN_BASE_URL
Gemini	create_gemini_llm_call()	GEMINI_API_KEY
Claude	create_claude_llm_call()	CLAUDE_CODE_KEY 或 ANTHROPIC_API_KEY，可选对应 Base URL
OpenRouter	create_openrouter_llm_call()	OPEN_ROUTER_API_KEY
Yescode	create_yescode_llm_call()	YESCODE_BASE_URL + YESCODE_API_KEY

RunConfig 关键参数

RunConfig(
    model="qwen3.5-plus",
    temperature=0.3,
    max_iterations=200,
    agent_type="default",           # Agent 预设（对应 presets.json）
    name="任务名称",
    tools=None,                      # 精确指定工具列表（优先于 tool_groups）
    tool_groups=["core", "browser"], # 工具分组白名单，默认 ["core"]
    goal_compression="on_overflow", # Goal 压缩："none" / "on_complete" / "on_overflow"
    knowledge=KnowledgeConfig(...), # 知识提取配置
)

续跑已有 Trace

config = RunConfig(model="qwen3.5-plus", trace_id="existing-trace-id")
async for item in runner.run(messages=[{"role": "user", "content": "继续"}], config=config):
    ...

Skill 指定注入（设计中）

同一个长期续跑的 agent，不同调用可以注入不同 skill：

# 不同调用场景注入不同策略
async for item in runner.run(
    messages=[...], config=config,
    inject_skills=["ask_strategy"],     # 本次需要的 skill
    skill_recency_threshold=10,         # 最近 N 条消息内有就不重复注入
):
    ...

详见 Context 管理 § Skill 指定注入

自定义工具

from cyber_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 预设（system prompt、skills 过滤）
├── requirement.prompt  # 任务 prompt（$system$ + $user$ 段）
├── skills/             # 项目自定义 skills
└── tools/              # 项目自定义工具

SDK 与 CLI 入口

入口	用途
AgentRunner.run() / run_result()	Python 底层 SDK，创建或续跑 Trace
cyber_agent.client.invoke_agent()	统一调用本地或 remote_* Agent
api_server.py	Trace REST API 和 WebSocket 服务
cyber_agent.cli.interactive.InteractiveController	示例运行器可接入的暂停、续跑、反思与压缩菜单
python -m cyber_agent.cli.extraction_review --trace <ID>	知识提取的审核和提交 CLI

---

文档

文档	内容
架构设计	框架完整架构
Agent 运行模式	Legacy / Recursive 模式、权限、调度、停止与资源预算
Recursive 任务协议	TaskBrief、TaskReport、Validator 和父级审核
Context 管理	注入机制、压缩策略、Skill 指定注入
工具系统	工具定义、注册、参数注入
Skills 指南	Skill 分类、编写、加载
Prompt 规范	Prompt 撰写原则
Trace API	REST API 和 WebSocket 接口
设计决策	架构决策记录

---

REST API

方法	路径	说明
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	停止运行

实现：cyber_agent/trace/api.py, cyber_agent/trace/run_api.py

---

Claude Code 集成

本仓库自带 Claude Code / Codex skill 源码：skills4claude/agent/

• SKILL.md — skill 元数据（description 给 Claude Code 路由用）
• invoke.py — 20 行薄脚本，from cyber_agent import invoke_agent 然后透传命令行参数

安装 skill

cyber-agent 包可 import 后，使用仓库安装脚本链接到 Claude Code：

pip install -e .
bash skills4claude/install.sh --claude --skills agent

invoke_agent 调用

Claude Code skill 脚本最终调用 cyber_agent/client.py::invoke_agent，该函数也是公开 SDK 供任何 Python 代码复用：

from cyber_agent import invoke_agent

# 远端：查询知识库
result = await invoke_agent(
    agent_type="remote_librarian",
    task="ControlNet 相关的工具知识",
    skills=["ask_strategy"],
)

# 本地：在项目目录下跑 agent（约定 project_root/config.py 定义 RUN_CONFIG）
result = await invoke_agent(
    agent_type="main",
    task="...",
    project_root="./examples/research",
)

路由规则：agent_type.startswith("remote_") → HTTP 调用 KnowHub /api/agent；否则在当前进程起 AgentRunner。

实现：cyber_agent/client.py:invoke_agent / skills4claude/agent/invoke.py

---

附录：工具分组

通过 RunConfig(tool_groups=[...]) 控制 Agent 可用的工具范围。默认仅 ["core"]。每个工具在 @tool(groups=[...]) 中声明分组，支持多标签。

机制详见 docs/tools.md § 工具分组。

core — 基础能力（13）

工具	说明
read_file	读取单个文件（文本/图片/PDF）
read_images	批量读图 + 网格拼图
edit_file	编辑文件
write_file	写入文件
glob_files	文件模式匹配
grep_content	内容搜索（正则）
bash_command	执行 shell 命令
skill	调用 skill
list_skills	列出可用 skill
agent	创建子 Agent
evaluate	评估执行结果
goal	目标/计划管理
get_current_context	获取当前执行上下文

browser — 浏览器自动化（14）

工具	说明
browser_navigate	导航到 URL
browser_search	搜索引擎搜索
browser_back	返回上一页
browser_interact	元素交互（click/type/send_keys/upload/dropdown）
browser_scroll	滚动页面
browser_screenshot	截图（可带元素编号标注）
browser_elements	获取可交互元素列表
browser_read	读取页面内容（html/find/long）
browser_extract	LLM 驱动的结构化数据提取
browser_tabs	标签页管理（switch/close）
browser_cookies	Cookie/登录态（load/export/ensure_login）
browser_wait	等待（定时/等用户操作）
browser_js	执行 JavaScript
browser_download	下载文件

content — 内容搜索（6）

工具	说明
content_platforms	列出/查询平台及参数（支持模糊匹配）
content_search	跨 11 平台搜索（小红书/B站/知乎/GitHub/YouTube/X 等）
content_detail	查看内容详情（从搜索缓存按索引取）
content_suggest	搜索关键词补全建议
extract_video_clip	YouTube 视频片段截取
import_content	批量导入文章到 CMS

远端 Agent — 统一通过 agent 工具

知识查询、知识上传、深度调研等远端服务不再作为独立工具暴露，全部通过统一的 agent 工具调用。Librarian 一个 Agent 多种模式，通过 skills 参数切换：

用法	说明
agent(agent_type="remote_librarian", task=..., skills=["ask_strategy"])	知识查询，返回带引用的整合回答
agent(agent_type="remote_librarian", task=json.dumps({...}), skills=["upload_strategy"])	知识上传（task 为 JSON 字符串）
agent(agent_type="remote_research", task=...)	深度调研

详见 tools.md § Agent 工具。

toolhub — 远程工具库（3）

工具	说明
toolhub_health	检查远程工具库状态
toolhub_search	搜索远程 AI 工具
toolhub_call	调用远程工具（图片参数支持本地路径）

feishu — 飞书（4）

工具	说明
feishu_get_contact_list	获取联系人列表
feishu_send_message_to_contact	给联系人发消息
feishu_get_contact_replies	获取联系人回复
feishu_get_chat_history	获取聊天历史

im — IM 通信（8）

工具	说明
im_setup	初始化 IM 连接
im_send_message	发送消息
im_receive_messages	接收消息
im_check_notification	检查通知
im_get_contacts	获取联系人
im_get_chat_history	获取聊天历史
im_open_window	打开 IM 窗口
im_close_window	关闭 IM 窗口

resource — 资源查询（2）

工具	说明
resource_list_tools	列出资源工具
resource_get_tool	获取工具详情

knowledge_internal — 知识库内部操作（14）

仅供 Librarian Agent 内部使用，普通 Agent 不可见。通过 tools=[...] 精确指定访问。

工具	说明
knowledge_search	知识检索（语义 + 精排）
knowledge_save	保存知识条目
knowledge_list	列出知识
knowledge_update	更新知识反馈
knowledge_batch_update	批量更新反馈
knowledge_slim	知识瘦身
resource_save	保存资源
resource_get	获取资源
tool_search	搜索工具记录
tool_list	列出工具记录
capability_search	搜索能力记录
capability_list	列出能力记录
requirement_search	搜索需求记录
requirement_list	列出需求记录