双系统认知架构的智能体框架——自动从成功任务中蒸馏方法论,在后续相似任务中零推理复用。
当前主流智能体框架每次任务都从零开始推理,产生极长上下文,严重消耗计算资源。人类解决问题采用双系统认知:
- 慢思考(System 2):遇到新问题,逐步试错探索
- 快思考(System 1):遇到熟悉问题,凭经验直觉执行
AgentIntelligence 将这种机制引入智能体系统:慢思考路径探索新任务并记录完整轨迹,反思模型自动蒸馏出可复用的方法论卡片;快思考路径匹配到已知方法论后,跳过 LLM 推理,用确定性代码直接执行。
pip install -e .
# 可选:本地嵌入模型(离线模式)
pip install -e ".[local-embed]"依赖:Python >= 3.10, pydantic, chromadb, openai, click, rich
# 设置 API key
export OPENAI_API_KEY="sk-..."
# 运行任务
agent-intelligence run "读取 config.json 并修改数据库连接字符串"
# 指定模型
agent-intelligence run "搜索最新的 Python 3.13 发布说明" --model gpt-4o
# 查看系统状态
agent-intelligence stats
# 搜索已有方法论
agent-intelligence search "文件操作"
# 运行维护清理
agent-intelligence maintain --dry-run用户任务 → 任务路由器(向量相似度匹配)
│
相似度 ≥ 阈值 相似度 < 阈值
│ │
快思考路径 慢思考路径
(确定性DAG执行) (Agentic探索循环)
│ │
零LLM调用 轨迹记录
│ │
直接调用工具 反思模型蒸馏
│ │
返回结果 生成方法论卡片 → 存入向量库
任务路由 — 计算任务与历史方法论的语义相似度,结合执行模型的参数规模,决定走快思考还是慢思考。
慢思考与蒸馏 — 对全新任务启动 Agentic 探索循环,反思模型分析完整轨迹,剔除失败分支,提取成功调用链,生成结构化方法论卡片及其多层级变体。
快思考执行 — 匹配到已知方法论后,从用户输入中提取参数槽位,按卡片定义的步骤 DAG 确定性调用工具,全程零 LLM 推理。
异常回退 — 快思考步骤失败时,打包已完成上下文退回慢思考重新探索;遗忘机制定期清理长期未使用或成功率低的方法论卡片。
一次成功任务的蒸馏结果,包含:
card = MethodologyCard(
name="发送邮件",
description="通过 SMTP 发送电子邮件",
parameter_slots=[
ParameterSlot(name="recipient", extraction_hints=[r"收件人[::]\s*(.+?)$"]),
ParameterSlot(name="subject", extraction_hints=[r"主题[::]\s*(.+?)$"]),
ParameterSlot(name="body", extraction_hints=[r"正文[::]\s*(.+?)$"]),
],
steps=[
ExecutionStep(order=1, tool_name="http_request", parameters={
"url": "https://api.mail.com/send",
"method": "POST",
"body": '{"to": "$slots.recipient", "subject": "$slots.subject"}',
}),
],
)系统定义 5 个能力层级,方法论卡片为每个层级生成适配变体:
| 层级 | 参数量 | 能力 | 方法论形式 |
|---|---|---|---|
| TINY | <3B | 模式匹配、模板填充 | 纯槽位填充 |
| SMALL | 3-14B | 基础推理、工具选择 | 模板 + 简单条件分支 |
| MEDIUM | 14-70B | 多步推理、歧义处理 | 完整步骤 + 条件逻辑 |
| LARGE | 70B+ | 复杂规划、新问题求解 | 完整方法论 + 边缘情况 |
| FRONTIER | 专有模型 | 全部能力 + 反思蒸馏 | 完整方法论 + 可进行蒸馏 |
蒸馏时自动为每个低层级生成降级变体:推理步骤转换为启发式规则(中型模型)、模板匹配(小型模型)、纯槽位提取(微型模型)。
from agentintelligence.agent import AgentIntelligence
from agentintelligence.llm.client import LLMClient, LLMConfig
ai = AgentIntelligence(
LLMClient(LLMConfig(
provider="openai",
model_name="gpt-4o",
)),
data_dir="./data",
)
# 执行任务(自动路由)
result = ai.run("读取 data.csv 并计算平均值")
print(f"Path: {result.path}") # fast 或 slow
print(f"Success: {result.success}")
print(f"LLM Calls: {result.total_llm_calls}") # fast 路径为 0
print(f"Latency: {result.total_latency_ms}ms")
# 查询统计
stats = ai.get_stats()
print(stats)
# 搜索方法论
cards = ai.search_methodologies("数据分析", top_k=5)
for card, score in cards:
print(f" {card.name} ({score:.3f})")
# 维护
report = ai.run_maintenance(dry_run=True)
ai.close()from agentintelligence.execution.tool_registry import ToolRegistry
registry = ToolRegistry()
registry.register(
name="send_email",
description="发送电子邮件",
callable=my_send_email_fn,
parameters={
"type": "object",
"properties": {
"to": {"type": "string"},
"subject": {"type": "string"},
"body": {"type": "string"},
},
"required": ["to", "subject", "body"],
},
required_params=["to", "subject", "body"],
)
ai = AgentIntelligence(tool_registry=registry)src/agentintelligence/
├── agent.py # 主协调器
├── models/ # 数据模型
│ ├── tier.py # 模型能力层级
│ ├── card.py # 方法论卡片
│ └── trajectory.py # 探索轨迹
├── storage/ # 双层存储
│ ├── vector_store.py # ChromaDB 向量检索
│ ├── doc_store.py # SQLite 结构化存储
│ └── repository.py # 统一仓储接口
├── routing/ # 任务路由
│ ├── router.py # 语义匹配与分流
│ └── embeddings.py # 嵌入生成
├── execution/ # 执行引擎
│ ├── fast_path.py # 快思考(零LLM)
│ ├── slow_path.py # 慢思考(探索循环)
│ ├── slot_extractor.py # 参数槽位提取
│ ├── step_validator.py # 输出验证
│ └── tool_registry.py # 工具注册
├── distillation/ # 轨迹蒸馏
│ ├── distiller.py # 蒸馏 → 方法论卡片
│ └── simplifier.py # 多层级变体生成
├── fallback/ # 异常处理
│ ├── manager.py # 回退管理
│ └── forgetting.py # 遗忘机制
├── llm/ # LLM 封装
│ ├── client.py # OpenAI / Anthropic
│ └── prompts.py # 提示词模板
└── cli/ # CLI
└── main.py
PYTHONPATH=src python -m pytest tests/ -v