|
|
||
|---|---|---|
| .agents/skills | ||
| .codegraph | ||
| .cursor/rules | ||
| .gitea/workflows | ||
| .github/workflows | ||
| .understand-anything | ||
| configs | ||
| docs | ||
| scripts | ||
| src/agentkit | ||
| src-tauri | ||
| test-results | ||
| tests | ||
| .dockerignore | ||
| .env.example | ||
| .env.test | ||
| .gitignore | ||
| AGENTS.md | ||
| CLAUDE.md | ||
| Dockerfile | ||
| README.md | ||
| agentkit.yaml | ||
| docker-compose.deploy.yaml | ||
| docker-compose.test.yml | ||
| docker-compose.yaml | ||
| ocr | ||
| pyproject.toml | ||
| skills-lock.json | ||
README.md
Fischer AgentKit
统一 AI Agent 开发框架 -- 将 LLM、Tool、Prompt 组装为可执行的 Skill,通过 ReAct 推理引擎自主完成任务,支持记忆持久化、自进化、Pipeline 编排和桌面客户端。
项目简介
AgentKit 解决的核心问题:从写 150 行 Agent 代码降为 10-20 行 YAML 配置。
传统方式下,每新增一个 Agent 需要编写子类、处理 LLM 调用、管理工具绑定、校验输出质量。AgentKit 将这些能力标准化为可组合模块,开发者只需编写 YAML 配置即可定义一个完整的 Skill(Prompt + Tool + 质量门禁),框架自动完成 ReAct 推理循环、模型路由降级、产出质量检查和标准化输出。
核心定位:
- 配置驱动 -- YAML 定义 Skill,无需写 Agent 子类
- 生产就绪 -- 内置质量门禁、模型降级、用量统计、级联检测、状态持久化
- 四种使用 -- Python 库引用、CLI 聊天、Web GUI、桌面客户端
- 专家团队 -- Expert Team Mode,多专家协作执行复杂任务,前端以多角色对话流呈现
- 记忆持久化 -- SOUL/USER/MEMORY/DAILY 四层记忆,写入即生效
- 自进化 -- 反思驱动 Soul 更新,经验积累与陷阱检测
- 工具丰富 -- 内置 Shell、搜索、爬虫、记忆、桌面操控等工具,支持 MCP 扩展
- Pipeline 编排 -- 多 Agent 协同、Saga 补偿、动态流水线
- 暗色主题 -- CSS 变量 + Ant Design 暗色算法,支持系统/手动切换
- LLM 缓存 -- 语义相似度缓存,减少重复调用,降低成本
核心特性
1. ReAct 推理引擎
Think -> Act -> Observe 循环。LLM 自主决定是否调用工具、调用哪个工具、何时给出最终答案。支持 Function Calling 和文本解析两种工具调用模式,最大步数可配置。
2. LLM Gateway
统一 LLM 调用入口。Provider 注册、模型别名解析(如 default -> dashscope/qwen3-coder-plus)、Fallback 降级策略、Token 用量和成本追踪。支持百炼 DashScope、OpenAI、DeepSeek 等 OpenAI 兼容 API。
3. Skill 系统
Skill = SkillConfig + 绑定 Tools。一个 Skill 代表一个可执行技能,包含 Prompt 模板、工具列表、意图配置和质量门禁。通过 YAML 配置即可定义,无需编写代码。
4. 意图路由
两级路由:Level 1 关键词匹配(零成本,~0ms),Level 2 LLM 分类(回退方案,~200 tokens)。自动将用户输入路由到最佳匹配的 Skill。
5. 记忆系统
四层持久化记忆,写入即生效(无需重启):
| 层级 | 文件 | 说明 |
|---|---|---|
| 身份 | SOUL.md |
Agent 身份、性格、做事准则、版本追踪 |
| 用户 | USER.md |
用户基本信息和偏好 |
| 笔记 | MEMORY.md |
Agent 主动记录的重要信息 |
| 日志 | DAILY/ |
按日期归档的交互摘要 |
- Section-based CRUD:每个记忆文件按
## Section组织,支持原子读写 - 容量保护:
trim_to_budget按 section 边界裁剪,保护"版本"和"更新历史" - 即时刷新:MemoryTool 写入后自动触发
notify_change(),所有 Agent 的 system_prompt 实时更新 - RAG 检索:向量嵌入 + 多源检索器,支持飞书/Confluence 适配器
6. 自进化系统
反思驱动的 Agent 自我改进:
- Reflector -- 任务完成后自动反思,生成质量评分和改进建议
- Soul Evolution -- 累积反思触发阈值后自动更新 SOUL.md,版本追踪
- 经验存储 -- 成功/失败经验持久化,陷阱检测避免重复错误
- Prompt 优化 -- 遗传算法 + A/B 测试自动优化 Prompt
- 路径优化 -- 分析工具调用路径,推荐更优执行策略
7. 三层意图路由
CostAwareRouter 三层路由,从零成本到高成本逐层升级:
| Layer | 方法 | 延迟 | Token 消耗 | 说明 |
|---|---|---|---|---|
| 0 | 正则规则 | ~0ms | 0 | 问候/简单对话/@team/@skill 前缀直接回复 |
| 1 | 启发式分类 | ~0ms | 0 | 关键词 + 模式匹配 + 复杂度评估 |
| 1.5 | 语义路由 | ~0ms | 0 | 向量相似度匹配(可选) |
| 2 | LLM 分类 | ~500ms | ~200 | 回退方案,LLM 判断意图 |
路由结果携带 ExecutionMode 枚举(DIRECT_CHAT / REACT / SKILL_REACT / TEAM_COLLAB),作为路由层与执行层的架构契约,杜绝硬编码。
8. 语义路由
基于向量相似度的意图路由,作为关键词匹配的补充:
- SemanticRouter -- 将用户输入和 Skill 描述向量化,通过余弦相似度匹配
- 缓存友好 -- 向量缓存避免重复计算
- 平滑降级 -- 语义路由失败时自动回退到启发式/LLM 分类
9. LLM 响应缓存
语义相似度缓存,减少重复 LLM 调用:
- CacheKey -- 基于 prompt + model + temperature 生成缓存键
- 语义匹配 -- 相似 prompt 命中缓存,避免重复调用
- TTL 管理 -- 缓存条目自动过期,支持手动失效
10. 级联检测与状态持久化
生产级故障防护:
- CascadeDetector -- 检测 Agent 输出中的级联失败模式,及时熔断
- CascadeStateStore -- 级联状态持久化,支持 InMemory 和 Redis 后端
- session_ttl -- 可配置的会话 TTL,自动清理过期状态
- 优雅降级 -- Redis 不可用时自动降级到 InMemory,保持服务可用
11. 产出质量管理
四维质量检查:必填字段、最低字数、JSON Schema 校验、自定义验证器。检查不通过时自动重试(可配置 max_retries),重试时携带质量反馈信息。
12. 标准化输出
Schema 验证 + 字段类型归一化(str -> int/float/bool)+ 元数据附加(version、produced_at、quality_score)。所有 Skill 产出统一为 StandardOutput 格式。
13. 内置工具集
开箱即用的工具插件,覆盖常见 Agent 需求:
| 工具 | 说明 |
|---|---|
ShellTool |
执行 Shell 命令,白名单安全机制 + 用户确认 |
WebSearchTool |
DuckDuckGo / Bing 网页搜索 |
BaiduSearchTool |
百度搜索 |
WebCrawlTool |
网页抓取与内容提取 |
MemoryTool |
短期/长期记忆管理 |
AskHumanTool |
向用户提问获取信息 |
SchemaExtractTool |
从文本提取结构化数据 |
SchemaGenerateTool |
生成 JSON Schema |
MCPTool |
MCP 协议工具扩展 |
ComputerUseTool |
桌面操控(截图、点击、输入),支持云端和本地(pyautogui)模式 |
工具组合:SequentialChain(顺序链)、ParallelFanOut(并行扇出)、DynamicSelector(动态选择)。
14. Pipeline 编排
多 Agent 协同编排,支持复杂工作流:
- PipelineEngine -- 阶段式流水线执行,支持自适应配置
- SagaOrchestrator -- 分布式事务补偿,失败自动回滚
- DynamicPipeline -- 运行时动态调整流水线结构
- PipelineReflector -- 执行反思与重规划
- HandoffManager -- Agent 间任务移交
15. Expert Team Mode
多专家协作执行复杂任务,B+C 混合模式(结构化协作计划 + 去中心化执行),前端以多角色对话流呈现:
核心组件:
| 组件 | 说明 |
|---|---|
ExpertConfig |
专家配置,扩展自 AgentConfig,新增 is_lead、expert_color、capabilities |
ExpertTemplate |
可复用的专家模板,通过 ExpertTemplateRegistry 管理 |
Expert |
专家实例,包装 ConfigDrivenAgent,支持 send_message、request_assist、handoff |
ExpertTeam |
团队容器,管理专家生命周期、共享工作区、协作计划 |
TeamOrchestrator |
计划执行引擎,支持串行/并行/竞争并行 + 结果合并 |
CollaborationPlan |
协作计划,定义阶段依赖、并行类型、合并策略 |
ExpertTeamRouter |
专家团队路由,@team 前缀触发,名称校验防注入 |
HandoffTransport |
专家间通信抽象,InProcess(asyncio.Queue)+ Redis Pub/Sub |
SharedWorkspace |
跨专家共享上下文,支持读写键值对 |
协作生命周期:
FORMING -> PLANNING -> EXECUTING -> SYNTHESIZING -> COMPLETED
|
失败时回退到单 Agent 模式
协作计划阶段类型:
| 类型 | 说明 | 合并策略 |
|---|---|---|
| 串行 | 按依赖顺序执行 | 最后阶段结果为最终结果 |
| 并行并行 | 多专家同时执行 | SEQUENTIAL / BEST / MERGE |
| 竞争并行 | 多专家竞争,选最优 | BEST(自动评分选择) |
前端对话 UI:
ExpertTeamView:专家头像列表 + 计划阶段进度条ExpertMessage:按专家角色渲染消息(头像、颜色、类型标签)PlanVisualization:协作计划时间线可视化- WebSocket 事件:
team_formed、expert_step、expert_result、plan_update、team_synthesis、team_dissolved
使用方式:
from agentkit.experts import ExpertConfig, ExpertTeam, ExpertTeamRouter
# 定义专家
researcher = ExpertConfig(name="researcher", is_lead=True, expert_color="#1890ff", ...)
writer = ExpertConfig(name="writer", expert_color="#52c41a", ...)
reviewer = ExpertConfig(name="reviewer", expert_color="#faad14", ...)
# 组建团队
team = ExpertTeam()
await team.form([researcher, writer, reviewer])
# 执行协作计划
from agentkit.experts import CollaborationPlan, PlanPhase, ParallelType
plan = CollaborationPlan(
id="plan-1", task="撰写深度分析报告", lead_expert="researcher",
phases=[
PlanPhase(id="p1", name="调研", assigned_expert="researcher", ...),
PlanPhase(id="p2", name="撰写", assigned_expert="writer", depends_on=["p1"], ...),
PlanPhase(id="p3", name="审校", assigned_expert="reviewer", depends_on=["p2"], ...),
],
)
team.update_plan(plan)
result = await orchestrator.execute_plan(plan)
用户也可在聊天中通过 @team:researcher,writer,reviewer 任务描述 前缀触发团队模式。
架构图
┌──────────────────────────────────────────────────────────────┐
│ 桌面客户端 (Tauri 2.x) │
│ splash → main窗口 → sidecar进程管理 → 系统托盘 │
└──────────────────────────┬───────────────────────────────────┘
│
┌──────────────────────────┼───────────────────────────────────┐
│ 前端 (Vue 3 + Ant Design Vue) │
│ ChatView · ExpertTeamView · ExpertMessage · PlanViz │
│ EvolutionView · WorkflowView · TerminalView · ComputerUse │
└──────────────────────────┼───────────────────────────────────┘
│ WebSocket / SSE / HTTP
┌──────────────────────────┼───────────────────────────────────┐
│ 服务端 (FastAPI + Uvicorn) │
│ portal.py · chat.py · evolution.py · workflows.py · ... │
│ 17个路由模块 · Agent Pool · Expert Team · Memory Store │
└──────────────────────────┼───────────────────────────────────┘
│
┌──────────────┼──────────────┐
│ CostAwareRouter │
│ Layer 0: 正则规则 (0ms) │
│ Layer 1: 启发式分类 (0ms) │
│ Layer 1.5: 语义路由 (可选) │
│ Layer 2: LLM分类 (~500ms) │
│ → ExecutionMode 枚举契约 │
└──────┬───────────────┬───────┘
│ │
DIRECT_CHAT │ │ REACT / SKILL_REACT / TEAM_COLLAB
▼ ▼
┌─────────────┐ ┌──────────────────┐ ┌──────────────────┐
│ Direct LLM │ │ ConfigDrivenAgent│ │ Expert Team │
│ (简单对话) │ │ (ReAct Engine) │ │ (多专家协作) │
└─────────────┘ └────────┬─────────┘ └────────┬─────────┘
│ │
┌────────────────┼────────────────┐ │
│ │ │ │
▼ ▼ ▼ ▼
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ LLM Gateway │ │ Tool Registry│ │ Memory System│
│ resolve→chat │ │ shell/search │ │ SOUL/USER │
│ fallback→track│ │ crawl/memory │ │ MEMORY/DAILY │
└──────┬───────┘ └──────────────┘ └──────────────┘
│
┌─────────┼─────────┐
▼ ▼ ▼
┌─────────┐ ┌────────┐ ┌──────────┐
│DashScope│ │ OpenAI │ │ DeepSeek │ ...
└─────────┘ └────────┘ └──────────┘
模块分层
| 层级 | 模块 | 说明 |
|---|---|---|
| API | server/ cli/ |
服务端路由 + 命令行入口 |
| Service | core/ chat/ skills/ experts/ |
Agent 引擎、路由、技能系统、专家团队 |
| Data | memory/ session/ bus/ |
记忆持久化、会话管理、消息总线 |
| Utility | llm/ tools/ evolution/ quality/ mcp/ |
LLM 网关、工具、进化、质量、MCP |
快速开始
安装
pip install fischer-agentkit
如需 MCP 支持:
pip install fischer-agentkit[mcp]
开发模式:
cd fischer-agentkit
pip install -e ".[dev]"
前置依赖
- Python >= 3.11
- Redis(可选,分布式模式需要)
- PostgreSQL + pgvector(可选,语义记忆需要)
CLI 快速开始
安装后即可使用 agentkit 命令行工具:
# 查看版本
agentkit version
# 初始化项目(生成配置文件)
agentkit init
# 启动 Web GUI 聊天界面(推荐)
agentkit gui --port 8002
# 启动 CLI 聊天
agentkit chat
# 启动 Server(API 模式)
agentkit serve --host 0.0.0.0 --port 8001
# 健康检查
agentkit doctor
# 提交任务(远程模式)
agentkit task submit --skill content_generator --input '{"topic": "AI趋势"}' --server-url http://localhost:8001
# 异步提交任务
agentkit task submit --skill content_generator --input '{"topic": "AI趋势"}' --mode async --server-url http://localhost:8001
# 查看任务状态
agentkit task status <task_id> --server-url http://localhost:8001
# 列出任务
agentkit task list --server-url http://localhost:8001
# 取消任务
agentkit task cancel <task_id> --server-url http://localhost:8001
# 列出已注册 Skill
agentkit skill list --server-url http://localhost:8001
# 加载 Skill 配置
agentkit skill load ./my_skill.yaml
# 查看 Skill 详情
agentkit skill info content_generator --server-url http://localhost:8001
# 查看 LLM 用量
agentkit usage --server-url http://localhost:8001
# 配对业务系统(生成 API Key 给业务系统使用)
agentkit pair --name geo-backend
# 输出: API Key + 连接指令
# 查看已配对的客户端
agentkit pair --list
# 撤销配对
agentkit pair --revoke geo-backend
# 也可以用 python -m 方式运行
python -m agentkit version
业务系统配对
业务系统(如 GEO)通过 agentkit pair 完成配对后,即可独立调用 AgentKit:
# 1. 在 AgentKit 服务器上执行配对
agentkit pair --name geo-backend --skills-dir ./configs/skills
# 2. 将输出的 API Key 配置到业务系统
# GEO 的 .env 文件:
AGENTKIT_SERVER_URL=http://agentkit:8001
AGENTKIT_API_KEY=ak_live_xxxxxxxxxxxx
# 3. 业务系统即可调用 AgentKit API
# POST http://agentkit:8001/api/v1/tasks
# Header: X-API-Key: ak_live_xxxxxxxxxxxx
配置优先级: 客户端自定义配置(pair 时指定)> init 默认配置 > 硬编码默认值
Docker 部署
# 初始化项目配置
agentkit init
# 编辑 .env 文件,填入 API Key
cp .env.example .env
# 编辑 .env ...
# 启动完整环境(AgentKit + Redis + PostgreSQL)
docker-compose up -d
# 查看日志
docker-compose logs -f agentkit
# 健康检查
docker-compose exec agentkit agentkit doctor
# 停止
docker-compose down
最小示例
import asyncio
from agentkit import LLMGateway, SkillConfig, Skill, ConfigDrivenAgent
from agentkit.llm.providers.openai import OpenAIProvider
async def main():
# 1. 初始化 LLM Gateway
gateway = LLMGateway()
gateway.register_provider("dashscope", OpenAIProvider(
api_key="sk-xxx",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
))
# 2. 定义 Skill
config = SkillConfig(
name="content_generator",
agent_type="content_generation",
description="内容生成 Skill",
task_mode="llm_generate",
prompt={
"identity": "你是一个专业的内容生成助手",
"instructions": "根据用户需求生成高质量内容",
"output_format": "以 JSON 格式输出",
},
llm={"model": "default", "temperature": 0.7},
execution_mode="react",
max_steps=5,
)
skill = Skill(config=config)
# 3. 创建 Agent 并执行任务
agent = ConfigDrivenAgent(config=config, llm_gateway=gateway)
await agent.start()
from agentkit.core.protocol import TaskMessage
from datetime import datetime, timezone
task = TaskMessage(
task_id="task-001",
agent_name="content_generator",
task_type="content_generation",
input_data={"topic": "AI 搜索引擎优化趋势"},
priority=0,
created_at=datetime.now(timezone.utc),
)
result = await agent.execute(task)
print(result.output_data)
await agent.stop()
asyncio.run(main())
部署方式
Import 模式
作为 Python 库直接引用,适合嵌入到现有项目中。
from agentkit import LLMGateway, SkillConfig, Skill, ConfigDrivenAgent
gateway = LLMGateway()
# ... 注册 provider、创建 skill、执行任务
Server 模式
FastAPI 独立部署,通过 HTTP API 调用。
# server.py
import uvicorn
from agentkit.server.app import create_app
from agentkit import LLMGateway
from agentkit.llm.providers.openai import OpenAIProvider
gateway = LLMGateway()
gateway.register_provider("dashscope", OpenAIProvider(
api_key="sk-xxx",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
))
app = create_app(llm_gateway=gateway)
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
启动:
python server.py
Docker 部署
# 启动完整环境(AgentKit + Redis + PostgreSQL + pgvector)
docker-compose up -d
# 查看日志
docker-compose logs -f agentkit
# 健康检查
docker-compose exec agentkit agentkit doctor
# 停止
docker-compose down
docker-compose.yaml 包含三个服务:
| 服务 | 镜像 | 端口 | 说明 |
|---|---|---|---|
agentkit |
自建 (python:3.11-slim) | 8001 | AgentKit 服务端 |
redis |
redis:7-alpine | 6379 | 消息总线 + 缓存 |
postgres |
pgvector/pgvector:pg15 | 5432 | 语义记忆向量存储 |
桌面客户端 (Tauri 2.x)
跨平台桌面应用,Rust Shell + Vue 3 前端 + Python Sidecar。
前置条件:Rust 工具链 + Node.js 18+ + Python 3.11+
# 1. 构建 Python sidecar
pip install pyinstaller
pyinstaller --onefile --name agentkit-server src/agentkit/__main__.py
# 2. 放置 sidecar(带平台后缀)
# macOS Apple Silicon:
cp dist/agentkit-server src-tauri/binaries/agentkit-server-aarch64-apple-darwin
# macOS Intel:
cp dist/agentkit-server src-tauri/binaries/agentkit-server-x86_64-apple-darwin
# Linux:
cp dist/agentkit-server src-tauri/binaries/agentkit-server-x86_64-unknown-linux-gnu
# Windows:
copy dist\agentkit-server.exe src-tauri\binaries\agentkit-server-x86_64-pc-windows-msvc.exe
# 3. 构建前端
cd src/agentkit/server/frontend
npm install
npm run build:frontend
# 4. 开发模式(热重载)
npm run tauri dev
# 5. 生产构建
npm run tauri build
# 产物:
# macOS: src-tauri/target/release/bundle/dmg/Fischer AgentKit.dmg
# macOS: src-tauri/target/release/bundle/macos/Fischer AgentKit.app
# Windows: src-tauri/target/release/bundle/msi/
# Linux: src-tauri/target/release/bundle/deb/
架构:
Tauri Shell (Rust)
├── 窗口管理(splash + main)
├── 系统托盘(显示窗口 / 退出)
└── Sidecar 进程管理
└── agentkit-server(PyInstaller 打包的 Python 服务端)
└── Uvicorn + FastAPI (--port 0 动态分配)
Tauri 启动时以 --port 0 启动 sidecar,解析 stdout 获取实际端口,前端通过该端口连接后端。
调用方式
Import 模式示例
import asyncio
from agentkit import (
LLMGateway, SkillConfig, Skill, ConfigDrivenAgent,
IntentRouter, QualityGate, OutputStandardizer,
)
from agentkit.llm.providers.openai import OpenAIProvider
from agentkit.core.protocol import TaskMessage
from datetime import datetime, timezone
async def main():
# 初始化 Gateway
gateway = LLMGateway()
gateway.register_provider("dashscope", OpenAIProvider(
api_key="sk-xxx", base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
))
# 定义多个 Skill
content_config = SkillConfig(
name="content_generator",
agent_type="content_generation",
task_mode="llm_generate",
prompt={
"identity": "你是内容生成助手",
"instructions": "生成 SEO 优化内容",
"output_format": "JSON: {content, word_count}",
},
llm={"model": "default"},
intent={
"keywords": ["生成", "内容", "写作"],
"description": "内容生成与写作",
"examples": ["帮我写一篇文章", "生成 SEO 内容"],
},
quality_gate={
"required_fields": ["content"],
"min_word_count": 100,
"max_retries": 2,
},
execution_mode="react",
max_steps=5,
)
optimizer_config = SkillConfig(
name="geo_optimizer",
agent_type="geo_optimization",
task_mode="llm_generate",
prompt={
"identity": "你是 GEO 优化专家",
"instructions": "优化内容以提升 AI 搜索可见性",
"output_format": "JSON: {optimized_content, seo_score, changes}",
},
llm={"model": "default"},
intent={
"keywords": ["优化", "GEO", "SEO"],
"description": "内容 GEO/SEO 优化",
"examples": ["优化这篇文章", "提升搜索排名"],
},
quality_gate={
"required_fields": ["optimized_content", "seo_score"],
"max_retries": 1,
},
execution_mode="react",
)
# 注册 Skill
from agentkit import SkillRegistry
registry = SkillRegistry()
registry.register(Skill(config=content_config))
registry.register(Skill(config=optimizer_config))
# 使用意图路由
router = IntentRouter(llm_gateway=gateway)
routing_result = await router.route(
input_data={"query": "帮我生成一篇关于 AI 的文章"},
skills=registry.list_skills(),
)
print(f"路由到: {routing_result.matched_skill} (method={routing_result.method}, confidence={routing_result.confidence})")
# 创建 Agent 并执行
matched_skill = registry.get(routing_result.matched_skill)
agent = ConfigDrivenAgent(config=matched_skill.config, llm_gateway=gateway)
await agent.start()
task = TaskMessage(
task_id="task-001",
agent_name=agent.name,
task_type=agent.agent_type,
input_data={"query": "帮我生成一篇关于 AI 的文章"},
priority=0,
created_at=datetime.now(timezone.utc),
)
result = await agent.execute(task)
# 质量检查
quality_gate = QualityGate()
quality_result = await quality_gate.validate(result.output_data or {}, matched_skill)
print(f"质量检查: {'通过' if quality_result.passed else '未通过'}")
# 标准化输出
standardizer = OutputStandardizer()
standard_output = await standardizer.standardize(
raw_output=result.output_data or {},
skill=matched_skill,
quality_result=quality_result,
)
print(f"标准化输出: skill={standard_output.skill_name}, quality_score={standard_output.metadata.quality_score}")
await agent.stop()
asyncio.run(main())
Server 模式示例
curl 调用
注册 Skill:
curl -X POST http://localhost:8000/api/v1/skills \
-H "Content-Type: application/json" \
-d '{
"config": {
"name": "content_generator",
"agent_type": "content_generation",
"task_mode": "llm_generate",
"description": "内容生成 Skill",
"prompt": {
"identity": "你是内容生成助手",
"instructions": "生成高质量内容",
"output_format": "JSON: {content, word_count}"
},
"llm": {"model": "default"},
"intent": {
"keywords": ["生成", "内容"],
"description": "内容生成"
},
"quality_gate": {
"required_fields": ["content"],
"min_word_count": 100,
"max_retries": 2
},
"execution_mode": "react"
}
}'
提交任务(指定 Skill):
curl -X POST http://localhost:8000/api/v1/tasks \
-H "Content-Type: application/json" \
-d '{
"skill_name": "content_generator",
"input_data": {"topic": "AI 搜索引擎优化趋势"}
}'
提交任务(意图路由自动匹配):
curl -X POST http://localhost:8000/api/v1/tasks \
-H "Content-Type: application/json" \
-d '{
"input_data": {"query": "帮我生成一篇文章"}
}'
创建 Agent:
curl -X POST http://localhost:8000/api/v1/agents \
-H "Content-Type: application/json" \
-d '{"skill_name": "content_generator"}'
查询 LLM 用量:
curl http://localhost:8000/api/v1/llm/usage
健康检查:
curl http://localhost:8000/api/v1/health
Python SDK 调用
import asyncio
from agentkit.server.client import AgentKitClient
async def main():
async with AgentKitClient("http://localhost:8000") as client:
# 注册 Skill
await client.register_skill({
"name": "content_generator",
"agent_type": "content_generation",
"task_mode": "llm_generate",
"prompt": {
"identity": "你是内容生成助手",
"instructions": "生成高质量内容",
"output_format": "JSON: {content, word_count}",
},
"llm": {"model": "default"},
"intent": {"keywords": ["生成", "内容"], "description": "内容生成"},
"quality_gate": {"required_fields": ["content"], "max_retries": 2},
"execution_mode": "react",
})
# 提交任务
result = await client.submit_task(
input_data={"topic": "AI 搜索引擎优化趋势"},
skill_name="content_generator",
)
print(result)
# 查询用量
usage = await client.get_usage()
print(usage)
asyncio.run(main())
Skill 配置 YAML 示例
name: content_generator
agent_type: content_generation
version: "1.0.0"
description: "AI 内容生成 Skill:支持选题推荐和文章生成"
task_mode: llm_generate
supported_tasks:
- generate_topics
- generate_article
max_concurrency: 2
input_schema:
type: object
required:
- target_keyword
properties:
target_keyword:
type: string
description: 目标关键词
brand_name:
type: string
description: 品牌名称
word_count:
type: integer
description: 目标字数
default: 2000
output_schema:
type: object
properties:
topics:
type: array
description: 选题列表
content:
type: string
description: 生成的文章内容
word_count:
type: integer
prompt:
identity: "你是一个专业的内容生成助手,擅长为品牌创作高质量的 SEO/GEO 优化内容"
context: "品牌需要通过优质内容提升在 AI 搜索引擎中的可见性"
instructions: |
根据用户提供的关键词和品牌信息,生成符合要求的内容。
- generate_topics: 生成选题列表
- generate_article: 生成完整文章
constraints: |
- 内容必须原创
- 关键词密度适中
- 文章结构清晰
output_format: "JSON: generate_topics 返回 {topics: [{title, reason, keywords}]},generate_article 返回 {content, word_count}"
llm:
model: "default"
temperature: 0.7
max_tokens: 4000
tools:
- retrieve_knowledge
intent:
keywords:
- 生成
- 内容
- 写作
- 文章
description: "内容生成与写作"
examples:
- "帮我写一篇文章"
- "生成 SEO 内容"
- "推荐选题"
quality_gate:
required_fields:
- content
min_word_count: 100
max_retries: 2
custom_validator: null
execution_mode: react
max_steps: 5
加载 YAML 配置:
from agentkit import SkillConfig, Skill
config = SkillConfig.from_yaml("configs/content_generator.yaml")
skill = Skill(config=config)
LLM 配置 YAML 示例
providers:
dashscope:
api_key: "${DASHSCOPE_API_KEY}"
base_url: "https://dashscope.aliyuncs.com/compatible-mode/v1"
models:
qwen3-coder-plus:
max_tokens: 64000
cost_per_1k_input: 0.00014
cost_per_1k_output: 0.00028
openai:
api_key: "${OPENAI_API_KEY}"
base_url: "https://api.openai.com/v1"
models:
gpt-4o:
cost_per_1k_input: 0.005
cost_per_1k_output: 0.015
model_aliases:
default: "dashscope/qwen3-coder-plus"
fast: "dashscope/qwen3-coder-plus"
powerful: "openai/gpt-4o"
fallbacks:
dashscope/qwen3-coder-plus:
- "openai/gpt-4o"
加载 LLM 配置:
from agentkit.llm.config import LLMConfig
from agentkit import LLMGateway
llm_config = LLMConfig.from_yaml("configs/llm.yaml")
gateway = LLMGateway(config=llm_config)
意图路由使用示例
from agentkit import IntentRouter, SkillRegistry, LLMGateway
gateway = LLMGateway()
# ... 注册 provider
registry = SkillRegistry()
# ... 注册多个 skill
router = IntentRouter(llm_gateway=gateway)
# 关键词匹配(零成本)
result = await router.route(
input_data={"query": "帮我生成一篇文章"},
skills=registry.list_skills(),
)
# result.matched_skill = "content_generator"
# result.method = "keyword"
# result.confidence = 1.0
# LLM 分类(关键词未命中时自动触发)
result = await router.route(
input_data={"query": "我想提升品牌在 AI 搜索中的表现"},
skills=registry.list_skills(),
)
# result.matched_skill = "geo_optimizer"
# result.method = "llm"
# result.confidence = 0.85
质量检查使用示例
from agentkit import QualityGate, Skill, SkillConfig
# 定义带质量门禁的 Skill
config = SkillConfig(
name="content_generator",
agent_type="content_generation",
task_mode="llm_generate",
prompt={"identity": "内容生成助手", "output_format": "JSON"},
quality_gate={
"required_fields": ["content", "word_count"],
"min_word_count": 200,
"max_retries": 3,
"custom_validator": "myapp.validators.content_quality_check",
},
)
skill = Skill(config=config)
# 执行质量检查
gate = QualityGate()
result = await gate.validate(
output={"content": "这是一篇短文", "word_count": 5},
skill=skill,
)
print(result.passed) # False(字数不足)
print(result.can_retry) # True(max_retries > 0)
for check in result.checks:
print(f" {check.name}: {'PASS' if check.passed else 'FAIL'} {check.message or ''}")
自定义验证器:
# myapp/validators.py
async def content_quality_check(output: dict) -> bool:
"""自定义质量验证器"""
content = output.get("content", "")
# 检查内容不含违禁词
forbidden = ["抄袭", "复制粘贴"]
return not any(word in content for word in forbidden)
模块详解
core/react -- ReAct 推理引擎
ReActEngine 实现 Think -> Act -> Observe 循环:
- Think: 将对话历史和工具 schema 发送给 LLM
- Act: 如果 LLM 返回 tool_calls,执行对应工具
- Observe: 将工具结果追加到对话历史,回到 Think
支持两种工具调用模式:
- Function Calling: LLM 原生返回
tool_calls(推荐) - 文本解析: 从 LLM 文本中提取
Action: tool_name(args)或```tool ```代码块
停止条件:LLM 不返回 tool_calls,或达到 max_steps。
危险工具确认流:非白名单命令触发 needs_confirmation,用户确认后以 _skip_dangerous_check=True 重新执行,避免无限循环。
chat/skill_routing -- CostAwareRouter 三层路由
三层路由从零成本到高成本逐层升级:
| Layer | 组件 | 延迟 | Token |
|---|---|---|---|
| 0 | RegexRules |
~0ms | 0 |
| 1 | HeuristicClassifier |
~0ms | 0 |
| 1.5 | SemanticRouter |
~0ms | 0 |
| 2 | LLMClassifier |
~500ms | ~200 |
路由结果包含 ExecutionMode 枚举(DIRECT_CHAT / REACT / SKILL_REACT / TEAM_COLLAB),作为路由层与执行层的架构契约。complexity 评分使用 if is not None 判断,避免 0.0 or default 误覆盖。@team:expert1,expert2 前缀直接路由到 TEAM_COLLAB 模式。
chat/semantic_router -- 语义路由
基于向量相似度的意图路由,作为关键词匹配的补充:
- SemanticRouter -- 将用户输入和 Skill 描述向量化,通过余弦相似度匹配
- 缓存友好 -- 向量缓存避免重复计算
- 平滑降级 -- 语义路由失败时自动回退到启发式/LLM 分类
llm/gateway -- LLM Gateway
统一 LLM 调用入口,核心能力:
- Provider 注册:
gateway.register_provider("openai", provider) - 模型别名:
"default"->"dashscope/qwen3-coder-plus" - Fallback 降级: 主模型失败时自动切换到备选模型
- 用量追踪: 按 agent_name、model 统计 Token 用量和成本
- 模型解析:
"provider/model"格式自动路由到对应 Provider - 响应缓存: 语义相似度缓存,减少重复调用(
llm/cache.py) - 用量存储: InMemory/Redis 双后端,支持分布式用量统计(
llm/providers/usage_store.py)
skills -- Skill 系统
Skill = SkillConfig + 绑定 Tools。SkillConfig 扩展自 AgentConfig,新增:
intent: 意图配置(关键词、描述、示例),供 IntentRouter 使用quality_gate: 质量门禁配置,供 QualityGate 使用execution_mode: 执行模式(react / direct / custom)max_steps: ReAct 最大步数
SkillRegistry 管理 Skill 的注册、发现、更新。
experts -- Expert Team Mode
多专家协作执行复杂任务,B+C 混合模式:
- ExpertConfig -- 扩展自 AgentConfig,新增
is_lead、expert_color、capabilities字段 - ExpertTemplate -- 可复用专家模板,通过
ExpertTemplateRegistry管理,支持 YAML 定义 - Expert -- 专家实例,包装 ConfigDrivenAgent,支持
send_message、request_assist、handoff操作 - ExpertTeam -- 团队容器,管理专家生命周期、SharedWorkspace、CollaborationPlan
- TeamOrchestrator -- 计划执行引擎,支持串行/并行/竞争并行,每阶段独立重试,失败级联标记,最终回退到单 Agent
- CollaborationPlan -- 协作计划,PlanPhase 定义依赖关系、并行类型、合并策略,
_phase_indexO(1) 查找,迭代 DFS 检测循环依赖 - ExpertTeamRouter --
@team:NAME前缀路由,名称正则校验防注入(^[a-zA-Z0-9_-]{1,64}$),最多 10 个专家 - HandoffTransport -- 专家间通信抽象,InProcessHandoffTransport(asyncio.Queue + sentinel 关闭)+ RedisHandoffTransport(Pub/Sub + 连接重置)
- SharedWorkspace -- 跨专家共享上下文,支持读写键值对
团队生命周期:FORMING -> PLANNING -> EXECUTING -> SYNTHESIZING -> COMPLETED。失败时自动回退到单 Agent 模式(lead 或首个活跃专家)。
router/intent -- 意图路由(已升级为 chat/skill_routing)
原两级路由已升级为 CostAwareRouter 三层路由(详见 chat/skill_routing 模块详解)。
quality/gate -- 产出质量管理
四维质量检查:
| 维度 | 配置字段 | 说明 |
|---|---|---|
| 必填字段 | required_fields |
检查 output 中是否包含指定字段且非 None |
| 最低字数 | min_word_count |
检查 output["content"] 的词数是否达标 |
| Schema 校验 | output_schema |
使用 jsonschema 校验 output 结构 |
| 自定义验证 | custom_validator |
点分路径导入的验证函数,支持同步/异步 |
检查不通过时,如果 max_retries > 0,BaseAgent.execute() 会自动重试,将质量反馈信息注入 quality_feedback 字段。
quality/cascade -- 级联检测与状态持久化
生产级故障防护:
- CascadeDetector -- 检测 Agent 输出中的级联失败模式(连续失败、质量退化),及时熔断
- CascadeStateStore -- 级联状态持久化,InMemory/Redis 双后端,支持
session_ttl配置 - AlignmentGuard -- 消息质量管控,集成在 MemoryBus 中
- 优雅降级 -- Redis 不可用时自动降级到 InMemory,保持服务可用
- close() 方法 -- 显式关闭 Redis 连接池,避免资源泄漏
quality/output -- 标准化输出
OutputStandardizer 将原始产出转换为 StandardOutput:
- Schema 验证(如 output_schema 存在)
- 字段类型归一化(str -> int/float/bool,根据 schema 定义)
- 附加元数据(version、produced_at、quality_score)
quality_score = 通过的检查数 / 总检查数。
core/base -- BaseAgent
所有 Agent 的基类,定义标准生命周期:
execute(task)为 final 方法,包含完整的计时、try/except、TaskResult 构建- 子类只需实现
handle_task(task) -> dict - 生命周期钩子:
on_task_start/on_task_complete/on_task_failed - 支持 Tool 插件、Memory 系统、LLM Gateway、Quality Gate 注入
- 分布式模式:通过 Redis 实现心跳、任务监听、Agent Handoff
core/config_driven -- ConfigDrivenAgent
配置驱动的 Agent,从 YAML/Dict 自动组装:
llm_generate: 渲染 Prompt -> 调用 LLM -> 解析 JSON 输出tool_call: 调用注册的 Tool 并返回结果custom: 自定义 handler 函数(点分路径动态导入)
v2 增强:接受 SkillConfig 时自动创建 Skill 并启用 ReAct 模式,Quality Gate 自动集成。
core/agent_pool -- AgentPool
运行时 Agent 实例池,管理 Agent 的创建、获取、删除。支持从已注册的 Skill 创建 Agent。
memory -- 记忆系统
四层持久化记忆,基于 Markdown section 的 CRUD 操作:
- MemoryFile -- 单个记忆文件(SOUL/USER/MEMORY/DAILY),支持
read_section/write_section/add_section/remove_section - MemoryStore -- 管理所有记忆文件,
build_system_prompt()将记忆注入 system_prompt - 即时刷新 --
notify_change()回调机制,MemoryTool 写入后自动刷新所有 Agent 的 system_prompt - 容量保护 --
trim_to_budget按 section 边界裁剪,protected_sections确保版本/更新历史不被裁剪 - 原子写入 --
_update_soul在内存中构建完整内容后一次性写入,避免先删后加导致数据丢失 - RAG -- 向量嵌入 + 多源检索器,支持飞书/Confluence 适配器
记忆注入格式:
<agent-identity>
## 身份
我是AK,一个专业的 AI 助手。
</agent-identity>
<user-profile>
## 基本信息
- 姓名:张三
</user-profile>
<agent-notes>
## 重要事项
...
</agent-notes>
<recent-activity>
## 2026-06-14
...
</recent-activity>
[base_prompt 行为指令]
evolution -- 自进化系统
反思驱动的 Agent 自我改进:
- Reflector -- 任务完成后自动反思,生成 quality_score 和 suggestions
- evolve_soul -- 累积反思达到阈值后触发 SOUL.md 更新,汇总所有反思建议(去重取 top 5)
- ExperienceStore -- 成功/失败经验持久化
- PitfallDetector -- 陷阱检测,避免重复错误
- PromptOptimizer -- 遗传算法优化 Prompt
- PathOptimizer -- 分析工具调用路径,推荐更优策略
- ABTester -- A/B 测试验证优化效果
bus -- 消息总线
进程内/跨进程消息传递:
- MemoryBus -- 进程内同步消息总线,集成 CascadeDetector 和 AlignmentGuard 进行消息质量管控
- RedisBus -- 基于 Redis Pub/Sub 的分布式消息总线,支持多实例部署
server -- FastAPI Server
独立部署模式,提供 RESTful API 和 Web GUI:
| 路径 | 方法 | 说明 |
|---|---|---|
/ |
GET | Web GUI 聊天界面 |
/ws/chat |
WebSocket | GUI 实时聊天通道 |
/api/v1/agents |
POST | 创建 Agent(指定 skill_name 或 config) |
/api/v1/agents |
GET | 列出所有 Agent |
/api/v1/agents/{name} |
GET | 获取 Agent 详情 |
/api/v1/agents/{name} |
DELETE | 删除 Agent |
/api/v1/tasks |
POST | 提交任务(支持意图路由) |
/api/v1/skills |
POST | 注册 Skill |
/api/v1/skills |
GET | 列出所有 Skill |
/api/v1/llm/usage |
GET | 查询 LLM 用量 |
/api/v1/health |
GET | 健康检查 |
Web GUI
通过 agentkit gui 启动,8 个页面视图:
| 视图 | 说明 |
|---|---|
| ChatView | 实时对话,WebSocket 流式传输,代码高亮,工具调用卡片,@-mention 技能推荐,Expert Team 协作视图 |
| EvolutionView | 自进化仪表盘,任务/经验/指标/优化面板 |
| WorkflowView | 工作流编辑器,Vue Flow 可视化编排 |
| TerminalView | 终端模拟器,PTY 会话 |
| KnowledgeBaseView | 知识库管理,文档上传/搜索/源配置 |
| SkillsView | 技能市场,技能卡片/详情 |
| SettingsView | 系统设置 |
| ComputerUseView | 计算机使用,桌面操控 |
暗色主题:支持亮色/暗色/跟随系统三种模式,CSS 变量 + Ant Design 暗色算法,通过导航栏灯泡图标切换,偏好持久化到 localStorage。
桌面客户端 (Tauri 2.x)
跨平台桌面应用,架构:
- Rust Shell -- 窗口管理(splash + main)、系统托盘、单实例锁
- Sidecar 管理 -- 以
--port 0启动 Python 后端,解析 stdout 获取动态端口 - 前端 -- Vue 3 SPA,通过动态端口连接后端
orchestrator -- Pipeline 编排
多 Agent 协同编排模块:
- PipelineEngine -- 按 Stage 定义顺序执行,支持自适应配置和反思重规划
- SagaOrchestrator -- 分布式事务补偿,失败步骤自动执行补偿操作
- DynamicPipeline -- 运行时根据条件动态调整流水线结构
- HandoffManager -- Agent 间任务移交,支持上下文传递
- PipelineStateMemory/Redis/PG -- 流水线状态持久化,支持内存、Redis、PostgreSQL 后端
配置参考
SkillConfig
继承自 AgentConfig,新增 v2 字段。
| 字段 | 类型 | 默认值 | 说明 |
|---|---|---|---|
name |
str | (必填) | Skill 名称,全局唯一标识 |
agent_type |
str | (必填) | Agent 类型 |
version |
str | "1.0.0" |
版本号 |
description |
str | "" |
描述 |
task_mode |
str | "llm_generate" |
任务模式:llm_generate / tool_call / custom |
supported_tasks |
list[str] | [agent_type] |
支持的任务类型列表 |
max_concurrency |
int | 1 |
最大并发数 |
input_schema |
dict | None | 输入 JSON Schema |
output_schema |
dict | None | 输出 JSON Schema |
prompt |
dict | None | Prompt 配置,包含 identity/context/instructions/constraints/output_format/examples |
llm |
dict | None | LLM 配置,包含 model/temperature/max_tokens |
tools |
list[str] | [] |
绑定的工具名称列表 |
memory |
dict | None | 记忆系统配置 |
custom_handler |
str | None | 自定义 handler 点分路径(custom 模式必填) |
intent |
dict | None | 意图配置(见 IntentConfig) |
quality_gate |
dict | None | 质量门禁配置(见 QualityGateConfig) |
execution_mode |
str | "react" |
执行模式:react / direct / custom |
max_steps |
int | 5 |
ReAct 最大步数 |
IntentConfig
| 字段 | 类型 | 默认值 | 说明 |
|---|---|---|---|
keywords |
list[str] | [] |
关键词列表,用于 Level 1 关键词匹配 |
description |
str | "" |
Skill 描述,用于 Level 2 LLM 分类 |
examples |
list[str] | [] |
示例输入,辅助 LLM 分类 |
QualityGateConfig
| 字段 | 类型 | 默认值 | 说明 |
|---|---|---|---|
required_fields |
list[str] | [] |
必填字段列表 |
min_word_count |
int | 0 |
最低字数要求(0 表示不检查) |
max_retries |
int | 0 |
质量检查不通过时的最大重试次数 |
custom_validator |
str | None | 自定义验证器的点分路径,如 myapp.validators.check |
LLMConfig
| 字段 | 类型 | 默认值 | 说明 |
|---|---|---|---|
providers |
dict[str, ProviderConfig] | {} |
Provider 配置,key 为 provider 名称 |
model_aliases |
dict[str, str] | {} |
模型别名映射,如 default: "dashscope/qwen3-coder-plus" |
fallbacks |
dict[str, list[str]] | {} |
降级策略,如 dashscope/qwen3-coder-plus: ["openai/gpt-4o"] |
ProviderConfig
| 字段 | 类型 | 默认值 | 说明 |
|---|---|---|---|
api_key |
str | "" |
API Key |
base_url |
str | "" |
API Base URL |
models |
dict[str, dict] | {} |
模型配置,key 为模型名,value 包含 cost_per_1k_input/cost_per_1k_output |
与 GEO 项目集成
Mode A: HTTP API 集成
GEO 后端通过 HTTP 调用 AgentKit Server,无需引入 Python 依赖。
+-------------------+ HTTP +-------------------+
| GEO Backend | --------------> | AgentKit Server |
| (FastAPI) | /api/v1/tasks | (FastAPI) |
+-------------------+ +-------------------+
集成步骤:
- 启动 AgentKit Server(独立进程或 Docker 容器)
# agentkit_server.py
import uvicorn
from agentkit.server.app import create_app
from agentkit import LLMGateway
from agentkit.llm.providers.openai import OpenAIProvider
gateway = LLMGateway()
gateway.register_provider("dashscope", OpenAIProvider(
api_key="sk-xxx",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
))
app = create_app(llm_gateway=gateway)
uvicorn.run(app, host="0.0.0.0", port=8001)
- 在 GEO 后端调用
# geo/backend/app/services/agentkit_client.py
import httpx
class AgentKitClient:
def __init__(self, base_url: str = "http://localhost:8001"):
self._client = httpx.AsyncClient(base_url=base_url)
async def submit_task(self, skill_name: str, input_data: dict) -> dict:
response = await self._client.post(
"/api/v1/tasks",
json={"skill_name": skill_name, "input_data": input_data},
)
response.raise_for_status()
return response.json()
async def register_skill(self, config: dict) -> dict:
response = await self._client.post(
"/api/v1/skills",
json={"config": config},
)
response.raise_for_status()
return response.json()
- 在 GEO 业务逻辑中使用
# geo/backend/app/services/content_service.py
from app.services.agentkit_client import AgentKitClient
agentkit = AgentKitClient()
async def generate_content(keyword: str, brand: str) -> dict:
result = await agentkit.submit_task(
skill_name="content_generator",
input_data={"target_keyword": keyword, "brand_name": brand},
)
return result["data"]
开发指南
项目结构
fischer-agentkit/
├── src/agentkit/ # Python 后端
│ ├── bus/ # 消息总线(MemoryBus + RedisBus)
│ ├── chat/ # 聊天路由(CostAwareRouter + ExecutionMode)
│ ├── cli/ # CLI 命令(Typer)
│ ├── core/ # 核心引擎(ReAct/Reflexion/ReWOO/ConfigDriven + HandoffTransport)
│ ├── evaluation/ # 评估系统(RAGAS)
│ ├── evolution/ # 自进化(反思/优化/陷阱检测/A/B测试)
│ ├── experts/ # 专家团队(Expert/Team/Orchestrator/Plan/Router/Config/Registry)
│ ├── llm/ # LLM 网关(6 Provider + 缓存 + 用量追踪)
│ ├── marketplace/ # 多Agent市场(拍卖/财富)
│ ├── mcp/ # MCP 协议
│ ├── memory/ # 记忆系统(SOUL/USER/MEMORY/DAILY + RAG)
│ ├── orchestrator/ # Pipeline 编排(Saga/动态流水线)
│ ├── org/ # 组织发现
│ ├── prompts/ # Prompt 模板
│ ├── quality/ # 质量保障(对齐/级联检测/门控)
│ ├── router/ # 意图路由
│ ├── server/ # FastAPI 服务端 + Vue 3 前端
│ ├── session/ # 会话管理
│ ├── skills/ # 技能系统
│ ├── telemetry/ # 遥测追踪
│ ├── tools/ # 工具插件(21个,含桌面操控)
│ └── utils/ # 工具函数
├── src-tauri/ # Tauri 桌面客户端(Rust)
│ ├── src/ # main.rs + lib.rs + sidecar.rs + tray.rs
│ └── binaries/ # Sidecar 二进制(平台相关)
├── configs/ # 配置文件(技能/LLM/GEO)
├── tests/ # 测试(unit + integration)
├── docs/ # 文档(brainstorms + plans)
├── Dockerfile # Docker 镜像构建
├── docker-compose.yaml # 生产编排
└── pyproject.toml # Python 项目配置
常用开发命令
# 后端
pip install -e ".[dev]" # 安装开发依赖
agentkit gui --port 8002 # 启动 Web GUI
agentkit serve --port 8001 # 启动 API 服务器
pytest # 运行全部测试
pytest -m "not integration" # 仅单元测试
pytest --cov=agentkit # 覆盖率
ruff check src/ && ruff format src/ # 代码检查和格式化
# 前端
cd src/agentkit/server/frontend
npm install # 安装依赖
npm run dev # Vite 开发服务器
npm run build:frontend # 生产构建
npm run typecheck # TypeScript 类型检查
# 桌面客户端
cd src/agentkit/server/frontend
npm run tauri dev # Tauri 开发模式
npm run tauri build # Tauri 生产构建
# Docker
docker-compose up -d # 启动完整环境
docker-compose logs -f agentkit # 查看日志
添加新 Skill
- 创建 YAML 配置文件
# configs/my_skill.yaml
name: my_skill
agent_type: my_task
task_mode: llm_generate
description: "我的自定义 Skill"
prompt:
identity: "你是 xxx 助手"
instructions: "执行 xxx 任务"
output_format: "JSON: {result}"
llm:
model: "deepseek"
temperature: 0.7
intent:
keywords: ["xxx", "yyy"]
description: "xxx 任务"
quality_gate:
required_fields: ["result"]
max_retries: 2
execution_mode: react
max_steps: 5
- 加载并使用
from agentkit import SkillConfig, Skill, SkillRegistry
config = SkillConfig.from_yaml("configs/my_skill.yaml")
skill = Skill(config=config)
registry.register(skill)
添加新 Tool
- 创建 Tool 类
# myapp/tools/search.py
from agentkit.tools.base import Tool
class SearchTool(Tool):
def __init__(self):
super().__init__(
name="search",
description="搜索知识库",
input_schema={
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"},
"top_k": {"type": "integer", "description": "返回数量", "default": 5},
},
"required": ["query"],
},
)
async def execute(self, *, query: str, top_k: int = 5) -> dict:
# 实现搜索逻辑
results = await do_search(query, top_k)
return {"results": results}
- 注册到 ToolRegistry
from agentkit.tools.registry import ToolRegistry
registry = ToolRegistry()
registry.register(SearchTool())
- 在 Skill 配置中引用
tools:
- search
ShellTool 安全机制:ShellTool 内置白名单(ls、cat、curl 等安全命令直接执行),非白名单命令会触发用户确认。在 GUI 中以确认卡片形式展示,用户点击"确认执行"后才运行。
代码风格
项目使用 Ruff 进行代码检查和格式化:
ruff check src/
ruff format src/
配置见 pyproject.toml 中的 [tool.ruff],目标 Python 3.11,行宽 100。