Fischer AgentKit

统一 Agent 开发框架 -- 将 LLM、Tool、Prompt 组装为可执行的 Skill，通过 ReAct 推理引擎自主完成任务。

项目简介

AgentKit 解决的核心问题：从写 150 行 Agent 代码降为 10-20 行 YAML 配置。

传统方式下，每新增一个 Agent 需要编写子类、处理 LLM 调用、管理工具绑定、校验输出质量。AgentKit 将这些能力标准化为 6 个可组合模块，开发者只需编写 YAML 配置即可定义一个完整的 Skill（Prompt + Tool + 质量门禁），框架自动完成 ReAct 推理循环、模型路由降级、产出质量检查和标准化输出。

核心定位：

• 配置驱动 -- YAML 定义 Skill，无需写 Agent 子类
• 生产就绪 -- 内置质量门禁、模型降级、用量统计
• 两种部署 -- Python 库直接引用，或 FastAPI 独立部署

核心特性

1. ReAct 推理引擎

Think -> Act -> Observe 循环。LLM 自主决定是否调用工具、调用哪个工具、何时给出最终答案。支持 Function Calling 和文本解析两种工具调用模式，最大步数可配置。

2. LLM Gateway

统一 LLM 调用入口。Provider 注册、模型别名解析（如 deepseek -> deepseek/deepseek-chat）、Fallback 降级策略、Token 用量和成本追踪。

3. Skill 系统

Skill = SkillConfig + 绑定 Tools。一个 Skill 代表一个可执行技能，包含 Prompt 模板、工具列表、意图配置和质量门禁。通过 YAML 配置即可定义，无需编写代码。

4. 意图路由

两级路由：Level 1 关键词匹配（零成本，~0ms），Level 2 LLM 分类（回退方案，~200 tokens）。自动将用户输入路由到最佳匹配的 Skill。

5. 产出质量管理

四维质量检查：必填字段、最低字数、JSON Schema 校验、自定义验证器。检查不通过时自动重试（可配置 max_retries），重试时携带质量反馈信息。

6. 标准化输出

Schema 验证 + 字段类型归一化（str -> int/float/bool）+ 元数据附加（version、produced_at、quality_score）。所有 Skill 产出统一为 StandardOutput 格式。

架构图

                         +------------------+
                         |   User Request   |
                         +--------+---------+
                                  |
                                  v
                    +-------------+--------------+
                    |        IntentRouter        |
                    |  (keyword -> LLM classify) |
                    +-------------+--------------+
                                  |
                          matched_skill
                                  |
                                  v
                    +-------------+--------------+
                    |       ConfigDrivenAgent     |
                    |  (SkillConfig-driven)       |
                    +-------------+--------------+
                                  |
                     +------------+------------+
                     |                         |
                     v                         v
           +---------+--------+     +----------+---------+
           |   ReActEngine     |     |  Traditional Mode  |
           | Think->Act->Observe|    | llm_generate/      |
           +---------+--------+     | tool_call/custom   |
                     |              +--------------------+
                     v
          +----------+----------+
          |     LLM Gateway     |
          |  resolve -> chat    |
          |  fallback -> track  |
          +----------+----------+
                     |
              +------+------+
              |             |
              v             v
        +-----+----+ +-----+-----+
        | Provider A| | Provider B|  ...
        +-----+----+ +-----+-----+
              |             |
              v             v
        +-----+----+ +-----+-----+
        |  Tool 1   | |  Tool 2   |  ...
        +-----------+ +-----------+

                     |
                     v
          +----------+----------+
          |    Quality Gate      |
          | required_fields      |
          | min_word_count       |
          | schema validation    |
          | custom validator     |
          +----------+----------+
                     |
                     v
          +----------+----------+
          |  OutputStandardizer  |
          | schema + normalize   |
          | + metadata           |
          +----------+----------+
                     |
                     v
              StandardOutput

快速开始

安装

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

# 启动 Server
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("openai", OpenAIProvider(
        api_key="sk-xxx",
        base_url="https://api.openai.com/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": "openai/gpt-4o", "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("openai", OpenAIProvider(
    api_key="sk-xxx",
    base_url="https://api.openai.com/v1",
))

app = create_app(llm_gateway=gateway)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

启动：

python server.py

调用方式

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("openai", OpenAIProvider(
        api_key="sk-xxx", base_url="https://api.openai.com/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": "openai/gpt-4o"},
        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": "openai/gpt-4o"},
        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": "openai/gpt-4o"},
      "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": "openai/gpt-4o"},
            "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: "deepseek"
  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:
  openai:
    api_key: "sk-xxx"
    base_url: "https://api.openai.com/v1"
    models:
      gpt-4o:
        cost_per_1k_input: 0.005
        cost_per_1k_output: 0.015
      gpt-4o-mini:
        cost_per_1k_input: 0.00015
        cost_per_1k_output: 0.0006
  deepseek:
    api_key: "sk-xxx"
    base_url: "https://api.deepseek.com/v1"
    models:
      deepseek-chat:
        cost_per_1k_input: 0.001
        cost_per_1k_output: 0.002

model_aliases:
  default: "deepseek/deepseek-chat"
  fast: "openai/gpt-4o-mini"
  powerful: "openai/gpt-4o"

fallbacks:
  openai/gpt-4o:
    - "deepseek/deepseek-chat"
  deepseek/deepseek-chat:
    - "openai/gpt-4o-mini"

加载 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 循环：

1. Think: 将对话历史和工具 schema 发送给 LLM
2. Act: 如果 LLM 返回 tool_calls，执行对应工具
3. Observe: 将工具结果追加到对话历史，回到 Think

支持两种工具调用模式：

• Function Calling: LLM 原生返回 tool_calls（推荐）
• 文本解析: 从 LLM 文本中提取 Action: tool_name(args) 或 ```tool ``` 代码块

停止条件：LLM 不返回 tool_calls，或达到 max_steps。

llm/gateway -- LLM Gateway

统一 LLM 调用入口，核心能力：

• Provider 注册: gateway.register_provider("openai", provider)
• 模型别名: "default" -> "deepseek/deepseek-chat"
• Fallback 降级: 主模型失败时自动切换到备选模型
• 用量追踪: 按 agent_name、model 统计 Token 用量和成本
• 模型解析: "provider/model" 格式自动路由到对应 Provider

skills -- Skill 系统

Skill = SkillConfig + 绑定 Tools。SkillConfig 扩展自 AgentConfig，新增：

• intent: 意图配置（关键词、描述、示例），供 IntentRouter 使用
• quality_gate: 质量门禁配置，供 QualityGate 使用
• execution_mode: 执行模式（react / direct / custom）
• max_steps: ReAct 最大步数

SkillRegistry 管理 Skill 的注册、发现、更新。

router/intent -- 意图路由

两级路由策略：

Level	方法	延迟	Token 消耗	置信度
1	关键词匹配	~0ms	0	1.0
2	LLM 分类	~500ms	~200	0.0-1.0

关键词匹配对 input_data 中所有字符串值（包括嵌套）进行大小写不敏感匹配。LLM 分类构建 prompt 列出所有 Skill 的名称、描述和示例，让 LLM 返回 JSON 格式的匹配结果。

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/output -- 标准化输出

OutputStandardizer 将原始产出转换为 StandardOutput：

1. Schema 验证（如 output_schema 存在）
2. 字段类型归一化（str -> int/float/bool，根据 schema 定义）
3. 附加元数据（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。

server -- FastAPI Server

独立部署模式，提供 RESTful API：

路径	方法	说明
/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	健康检查

配置参考

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: "deepseek/deepseek-chat"
fallbacks	dict[str, list[str]]	{}	降级策略，如 openai/gpt-4o: ["deepseek/deepseek-chat"]

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)        |
+-------------------+                   +-------------------+

集成步骤：

1. 启动 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("deepseek", OpenAIProvider(
    api_key="sk-xxx",
    base_url="https://api.deepseek.com/v1",
))

app = create_app(llm_gateway=gateway)
uvicorn.run(app, host="0.0.0.0", port=8001)

2. 在 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()

3. 在 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"]

开发指南

运行测试

# 安装开发依赖
pip install -e ".[dev]"

# 运行全部测试
pytest

# 运行单元测试（跳过集成测试）
pytest -m "not integration"

# 运行并查看覆盖率
pytest --cov=agentkit --cov-report=term-missing

# 仅运行 Redis 相关测试
pytest -m redis

# 仅运行 PostgreSQL 相关测试
pytest -m postgres

添加新 Skill

1. 创建 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

2. 加载并使用

from agentkit import SkillConfig, Skill, SkillRegistry

config = SkillConfig.from_yaml("configs/my_skill.yaml")
skill = Skill(config=config)
registry.register(skill)

添加新 Tool

1. 创建 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}

2. 注册到 ToolRegistry

from agentkit.tools.registry import ToolRegistry

registry = ToolRegistry()
registry.register(SearchTool())

3. 在 Skill 配置中引用

tools:
  - search

代码风格

项目使用 Ruff 进行代码检查和格式化：

ruff check src/
ruff format src/

配置见 pyproject.toml 中的 [tool.ruff]，目标 Python 3.11，行宽 100。