一个不依赖任何 Agent 框架(LangChain / LangGraph等)的最小可用 Agent 实现。 使用 DeepSeek API 作为 LLM 后端,支持工具调用、多会话管理、上下文压缩。
pip install -r requirements.txtcp .env.example .env
# 编辑 .env,填入你的 DeepSeek API Keypython main.py[default] > 帮我算一下 (123 + 456) * 789
Agent: (123 + 456) × 789 = 457,047
[default] > 帮我创建一个待办:完成周报
Agent: 已为你创建待办 #1: 完成周报
[default] > 搜索一下最近的AI新闻
Agent: 搜索到以下结果:...
[default] > /new
New session: a3f2c1e0
[a3f2c1e0] > 你好,我是新会话
Agent: 你好!...
[a3f2c1e0] > /switch default
Switched -> session default
pytest tests/ -v| 命令 | 说明 |
|---|---|
/sessions |
列出所有会话 |
/switch <id> |
切换到指定会话 |
/new |
创建新会话 |
/trace |
查看当前会话的消息记录 |
/stats |
查看上下文统计信息 |
/clear |
清空当前上下文 |
/quit |
退出 |
用户 --> CLI (main.py) --> SessionManager --> Agent Loop --> LLM (DeepSeek)
| |
Session 1 ToolRegistry
Session 2 |- calculator
... |- search
|- todo
| 组件 | 文件 | 职责 |
|---|---|---|
| Agent Loop | agent.py |
核心循环:接收输入 -> 调 LLM -> 执行工具 -> 循环或返回 |
| LLM Client | llm.py |
封装 DeepSeek API,兼容 OpenAI SDK |
| Tool Registry | tools/registry.py |
工具注册、Schema 导出、按名执行 |
| Context Manager | context.py |
消息存储、构建 LLM 输入、上下文压缩 |
| Session Manager | session.py |
多会话隔离、生命周期管理 |
| Config | config.py |
环境变量和默认配置 |
Agent Loop 是整个系统的核心,每一轮循环执行以下步骤:
1. 从 ContextManager 构建 messages 列表
2. 调用 LLM(附带所有工具的 JSON Schema)
3. 解析 LLM 响应:
|- 如果 response 包含 tool_calls:
| a. 将 assistant 消息(含 tool_calls)加入 context
| b. 逐个执行工具,获取结果
| c. 将工具结果作为 tool 消息加入 context
| d. continue -> 回到步骤 1(让 LLM 看到工具结果)
|- 如果 response 是纯文本:
a. 将 assistant 消息加入 context
b. 返回文本给用户 -> 结束
关键设计决策:
- LLM 自主决策:通过 OpenAI function-calling 协议,LLM 根据工具 Schema 自行判断是否需要调用工具、调用哪个工具、传什么参数
- 多工具并行:如果 LLM 一次返回多个 tool_calls,按顺序执行后一起送回 LLM
- 循环上限:max_turns 防止无限循环(默认 10 轮)
每个工具继承 Tool 基类,定义三个属性和一个方法:
class MyTool(Tool):
name = "tool_name" # 唯一标识,LLM 通过名字调用
description = "工具描述" # LLM 看到的说明
parameters = { ... } # JSON Schema,LLM 据此构造参数
def execute(self, **kwargs): # 实际执行逻辑
return "结果字符串"ToolRegistry 负责:
register(tool)— 注册工具,检查名称唯一性get_schemas()— 导出所有工具的 OpenAI function-calling 格式execute(name, args)— 按名称查找并执行,返回字符串结果
SessionManager 用字典维护多个独立 Session,每个 Session 拥有独立的 ContextManager。
sessions = SessionManager()
s1 = sessions.get_or_create("window-1") # 窗口 1
s2 = sessions.get_or_create("window-2") # 窗口 2
# s1 和 s2 的上下文完全隔离
agent.run("查天气", s1.context, s1.id)
agent.run("写周报", s2.context, s2.id)
# 互不影响ContextManager 维护的消息结构:
[0] system prompt 始终保留
[1] compressed_summary (可选) 历史摘要
[2..N] recent_messages 完整保留
|- user messages
|- assistant messages (含 tool_calls)
|- tool result messages
压缩触发条件:len(messages) > max_messages(默认 20 条)
压缩方式:将超出部分的消息合并为一段摘要文本,截断到 2000 字符以内。多次压缩时,新摘要追加到旧摘要后面。
Memory(上下文记忆)的召回发生在 每次调用 LLM 之前,具体是 Agent.run() 中 context.get_messages() 被调用时。
这意味着:
- 每一轮 Agent Loop 都会自动召回全部上下文
- 工具执行后再次调用 LLM 时,工具结果也已在上下文中
- 追问(follow-up)天然支持,因为之前的对话都在 context 里
按以下顺序构建 messages 列表:
| 位置 | 角色 | 内容 | 说明 |
|---|---|---|---|
| [0] | system | 系统提示词 | 定义角色、规则、工具使用方式 |
| [1] | assistant | 压缩摘要 | 历史对话的压缩文本(如有) |
| [2+] | user/assistant/tool | 完整消息 | 最近 N 条对话的详细记录 |
| 信息类型 | 是否放入 | 理由 |
|---|---|---|
| 用户输入 | 完整放入 | LLM 需要理解用户意图 |
| Agent 文本回复 | 完整放入 | 保持对话连贯性 |
| 工具调用请求 | 完整放入(tool_calls) | LLM 需要知道自己调过什么工具 |
| 工具执行结果 | 完整放入 | LLM 需要工具结果来生成回复 |
| 系统提示词 | 始终保留 | Agent 行为基准 |
| 超过 20 条的旧消息 | 压缩为摘要 | 节省 token,保留关键信息 |
纯对话追问(如"还有呢?"):用户的新消息自动加入 context,LLM 能看到之前所有对话,自然理解"还有呢"指的是什么。
带工具的追问(如"再加一个待办"):同上,LLM 看到之前的待办操作记录,理解"再加一个"的上下文,会再次调用 todo 工具。
- 工具创造:AI 帮忙创造了所有工具类
- 上下文压缩策略:AI 帮助实现了"摘要 + 保留最近 N 条"的方案
- 测试用例:AI 辅助设计 mock LLM 响应的方式,用 unittest.mock 测试 Agent Loop
问题 1:如何解析 LLM 的工具调用?
- 初始想法:自己定义 JSON 格式让 LLM 输出,然后正则解析
- 最终方案:使用 OpenAI 的 function-calling 协议,DeepSeek API 原生支持,解析更可靠
问题 2:多个用户并发怎么办?
- 初始想法:用线程锁保护共享状态
- 最终方案:每个 Session 完全独立(独立 ContextManager),无共享状态,天然支持并发
问题 3:上下文太长怎么办?
- 初始想法:简单截断,丢弃最早的消息
- 最终方案:压缩为摘要 + 保留最近消息,摘要放在 system 后面,让 LLM 有"记忆"感
问题 4:工具执行失败怎么处理?
- 方案:工具返回错误字符串(而非抛异常),LLM 能看到错误信息并告知用户
问题 5:如何避免无限循环?
- 方案:max_turns 计数器,每轮循环 +1,超限后强制返回当前结果