一个基于 DeepAgents + FastAPI + DeepSeek 的 AI 工程学习助手 MVP。
本项目是练手学习项目:市面上基于 DeepAgents 的开源项目太少,国内相关课程又基本都要收费,于是做一个免费、可跑通、代码带中文注释的最小实现,供大家学习 DeepAgents 的工程组织方式。
| 组件 | 选型 | 作用 |
|---|---|---|
| Web 层 | FastAPI + SSE | 极简聊天页面、流式事件推送(Debug Timeline) |
| Agent runtime | DeepAgents | 主 Agent 每轮新建一个 graph,跑 ReAct 循环 |
| 模型 | DeepSeek(deepseek-v4 系列) | 支持 thinking 思考模式,1M 上下文 |
| 持久化 | SQLite + SQLModel | 会话 / 消息 / 会话摘要,无向量库 |
| 上下文管理 | 自研 | 固定前缀(kv cache 友好)+ 80% 阈值提炼式压缩 + skills 动态加载 |
| 前端 | 原生 HTML/CSS/JS | 无框架,便于阅读 |
learning_assistant/
├── agent/ # 主 Agent 装配与运行
│ ├── main_agent.py # MainAgent:每轮建 graph、事件流转换、fatal 兜底
│ ├── turn_builder.py # 消息拼装:静态区 + 摘要 + 边界后历史 + 当前消息
│ ├── event_translator.py # DeepAgents 原生事件 → 项目 StreamEvent
│ ├── compactor.py # 上下文压缩(触发判断 / 写盘归档 / 提炼式摘要)
│ ├── prompts/ # 所有 prompt(Markdown 文件,代码零硬编码)
│ ├── skills/ # 技能目录(SKILL.md + frontmatter)
│ ├── subagents.py # 角色型子代理(概念讲解 / 练习教练 / 项目导师)
│ └── tools/ # 内部 Python 工具(memory / sandbox / skills / registry / 错误分类)
├── web/ # FastAPI 层
│ ├── routes.py # HTTP + SSE 路由
│ ├── chat_service.py # 回合编排(持久化边界、压缩触发)
│ ├── dependencies.py # 模型 / 凭据装配工厂(配置文件)
│ └── static/ # 前端(原生 JS)
├── memory/ # SQLite 持久化(models + service)
├── streaming/ # 结构化事件 + SSE 编码
├── integrations/mcp/ # 外部 MCP server 配置加载
├── core/ # settings(.env 配置)
└── db/ # 数据库 engine / session
config/ # 运行时配置(mcp.servers.json)
data/ # SQLite 数据文件
sandbox/ # 学习产物目录(Agent 写入的唯一位置)
用户发消息
→ chat_service 落库 user 消息
→ 压缩触发判断(上一轮 input_tokens > 窗口 × 80%?是则先压缩)
→ turn_builder 拼装消息(system 静态区 + 摘要 + 边界后历史 + 当前消息)
→ MainAgent 每轮新建 DeepAgents graph 并执行
→ DeepAgents 原生事件流 → event_translator 转成项目 StreamEvent
→ SSE 推给前端(思考 / 活动 / 回答增量 / token 用量)
→ 工具调用(TOOL_CALL / TOOL_RESULT)落库
→ 结果落库 + DONE 事件
前置要求:
- Python 3.12+
- uv(本项目用 uv 管理环境,
uv run自动使用.venv)
# 1. 同步依赖
uv sync
# 2. 创建配置(至少填 DEEPSEEK_API_KEY)
cp .env.example .env
# 然后编辑 .env,填入你的 DeepSeek API Key
# 3. 启动开发服务器
uv run uvicorn learning_assistant.web.app:app --reload
# 4. 打开浏览器访问
# http://127.0.0.1:8000首次启动会自动建表(
data/app.db)。存量旧库的表结构升级见scripts/migrate_context_columns.py(幂等迁移脚本)。
所有配置通过根目录 .env 注入(learning_assistant/core/settings.py),
完整模板见 .env.example。
| 变量 | 说明 | 默认值 | 必填 |
|---|---|---|---|
DEEPSEEK_API_KEY |
DeepSeek API Key(platform.deepseek.com 获取) | 无 | ✅ |
DEEPSEEK_BASE_URL |
DeepSeek API 地址 | https://api.deepseek.com |
|
DEEPSEEK_MODEL |
模型名(deepseek-v4 系列,1M 上下文) | deepseek-v4-flash |
|
DEEPSEEK_THINKING_ENABLED |
是否开启思考模式 | true |
|
DEEPSEEK_REASONING_EFFORT |
思考强度:low / medium / high |
high |
|
DEEPSEEK_MAX_TOKENS |
单次回复最大输出 token(含思考内容,太小长思考会被截断) | 16384 |
|
DATABASE_URL |
SQLite 连接串 | sqlite:///./data/app.db |
|
MCP_CONFIG_PATH |
MCP server 配置文件路径 | ./config/mcp.servers.json |
|
SANDBOX_API_URL |
远程沙箱 API 地址(不配则自动禁用 sandbox 工具) | 无 | |
SANDBOX_API_TOKEN |
远程沙箱 API Token | 无 | |
CONTEXT_WINDOW_TOKENS |
模型上下文窗口(压缩触发阈值 = 窗口 × 80%) | 1000000 |
|
CONTEXT_ARCHIVE_DIR |
压缩归档目录(按会话 id 分目录,只存工具消息) | ./learning_assistant/messages |
安全提醒:
.env已被.gitignore忽略,请勿手动提交真实 Key; 公开仓库请始终通过.env.example分享配置模板。
沙箱是一个远程 HTTP API,本项目不内置沙箱服务器实现,需自备 (或用任意能提供下列端点的服务)。
配置后 Agent 获得 3 个工具;不配置则自动跳过(工具列表中不会出现)。
SANDBOX_API_URL=https://your-sandbox.example.com
SANDBOX_API_TOKEN=your-token请求使用 Authorization: Bearer <token> 认证,约定 3 个端点:
| 端点 | 方法 | 请求体 / 参数 | 响应 |
|---|---|---|---|
/execute |
POST | {"code": "..."} |
{"exit_code": 0, "stdout": "...", "stderr": ""} |
/files/write |
POST | {"path": "相对路径", "content": "..."} |
{"content": "..."} |
/files/read |
GET | ?path=相对路径 |
{"content": "..."} |
约定:只要 HTTP 返回 2xx 就算"调用成功",即使
exit_code != 0(用户代码运行失败会原样返回给模型,让模型自己修正);只有真正的 HTTP 错误(非 2xx / 超时 / 连接失败)才会抛异常并转为[TOOL_ERROR]失败 Observation。路径校验:
path必须是沙箱内相对路径,禁止绝对路径与..穿越。
Skills 表示"方法论"(怎么讲概念、怎么出练习、怎么计划复习),是 DeepAgents 原生 Markdown 技能,本项目自研加载(不用框架自带的 SkillsMiddleware / read_file 指引)。
learning_assistant/agent/skills/
├── concept_explainer/SKILL.md # 概念讲解方法论
├── exercise_generator/SKILL.md # 练习生成方法论
└── review_planner/SKILL.md # 复习计划方法论
---
name: concept_explainer
description: 用于解释编程、AI 工程和工具链概念的中文学习辅导方法。
---
# Concept Explainer Skill
正文(完整方法论)...
⚠️ ---与name:/description:必须顶格(第一列),带缩进会 导致 YAML 解析失败、该技能被静默跳过。
- 元数据(name / description)由
turn_builder扫描注入 system prompt 的「可用技能」区块(静态区,字节级稳定,kv cache 友好); - 模型需要某技能全文时调用
load_skill(name)工具——返回 SKILL.md 全文作为工具结果追加到消息末尾(不污染静态区); load_skill带路径白名单(name 必须来自扫描结果),天然防目录穿越;- 每次调用实时扫描:新增 / 修改技能无需重启,即时生效。
外部能力通过 MCP(Model Context Protocol)接入,配置文件默认在
config/mcp.servers.json(路径由 MCP_CONFIG_PATH 指定)。
{
"servers": {
"my-server": {
"transport": "http",
"url": "http://127.0.0.1:8080/mcp",
"headers": {},
"timeout": 30
}
}
}支持的 transport:
| transport | 必填字段 | 可选字段 |
|---|---|---|
stdio |
command |
args / env / cwd |
http / streamable_http |
url |
headers / timeout / sse_read_timeout |
注意:
config/目录会被 git 跟踪,不要把真实 Token 写进mcp.servers.json;/api/debug/tools调试接口对 MCP 做了容错—— 服务器不可达时只展示静态工具名,不会报错。
本项目自研了一套"固定前缀 + 提炼式压缩 + skills 动态加载"的上下文管理 (不依赖 LangGraph checkpointer,显式持久化,便于学习每一步):
- 固定前缀(kv cache 友好):system prompt(主指令 + AGENTS.md + skills 元数据)字节级稳定,绝不随会话变化;摘要作为 user 消息放在 其后(而不是塞进 system prompt),服务端缓存可命中。
- 工具调用可回放:
TOOL_CALL/TOOL_RESULT结构化落库,历史中的 assistant 消息携带tool_calls、tool 消息携带tool_call_id,可完整回放。 - 80% 阈值压缩:上一轮
input_tokens > 窗口 × 80%时,下一轮开始前 用主模型增量式提炼新摘要(旧摘要 + 被压缩消息 → 新摘要),只保留 任务目的 / 进度 / 工具结论,闲聊舍弃;被压缩部分的工具消息写盘归档 ({archive_dir}/{conversation_id}/compact.json);保留最近 3 轮原样。 - skills 动态加载:元数据常驻静态区、全文按需
load_skill(见上文)。
uv run python -m pytest当前 175 个测试全部通过(不依赖真实 LLM:配置、memory、MCP 解析、 SSE 编码、事件转换、拼装、压缩、工具装配等关键逻辑均有单测)。
# 只跑某一个模块
uv run python -m pytest tests/agent/test_compactor.py -v这个项目是学习用最小实现,明确不做(也请勿当作生产方案):
- 单用户、无登录 / 认证 / 多租户
- 无向量库 / 语义检索(会话历史靠 SQL 查询)
- 无数据库迁移工具(Alembic),表结构变更需手动迁移脚本
- 模型 / API / streaming 失败不做自动 fallback(显式报错)
- 不使用 LangGraph checkpointer(显式持久化 + 每轮重建消息,便于观察)
- 压缩触发阈值为硬编码 80%(
should_compact) - 无多模态、文件上传、语音
- 前端为原生 JS,界面朴素;无 WebSocket(SSE 单向推送)
- 不内置沙箱服务器实现(需自备远程沙箱 API)
按依赖顺序阅读代码,事半功倍:
- 装配:
web/dependencies.py(工厂)→agent/main_agent.py(引擎,每轮建 graph) - 工具:
agent/tools/registry.py(工具装配 seam)→tools/skill_loader/tools/sandbox_tools.py/tools/tool_errors.py(错误分类与安全包装) - 事件流:
streaming/events.py(事件契约)→agent/event_translator.py(原生事件 → StreamEvent)→web/chat_service.py(落库 / 持久化边界) - 上下文管理:
agent/turn_builder.py(拼装)→agent/compactor.py(压缩) - 持久化:
memory/models.py→memory/service.py
每个模块都带中文注释说明"为什么这么设计",可配合 Debug Timeline (前端右侧面板)观察每一轮的工具调用与 token 消耗。