Skip to content

About

面向复杂技术任务的可动态重规划 Agent:任务拆解 → 工具执行 → 状态评估 → 动态重规划 → 断点恢复 → 人工审批。LangGraph + DeepSeek + FastAPI/SSE + PostgreSQL + Docker,附 100 条评测集 Fixed vs Adaptive 对照实验

Topics

Resources

Stars

0 stars

Watchers

0 watching

Forks

Latest commit

 

History

44 Commits

Folders and files

Repository files navigation

Adaptive Task Planning Agent

面向复杂技术任务的可动态重规划 Agent:任务拆解 → 工具执行 → 状态评估 → 动态重规划 → 断点恢复 → 人工审批,附带 100 条评测集的 Fixed vs Adaptive 对照实验。

技术栈:Python · LangGraph · DeepSeek(OpenAI 兼容)· Pydantic · SQLite/PostgreSQL(Checkpointer)· prompt_toolkit · pytest

它解决什么问题

普通 Agent 是"计划一次、闷头执行到底"。本系统的核心机制是计划可以被执行证据修改:

  • Planner 用结构化输出把模糊目标拆成带工具绑定的任务清单;
  • Executor 逐个调用工具,瞬时故障自动指数退避重试;
  • Evaluator 判定每个任务是否"真正达成意图"(工具没报错 ≠ 成功);
  • Replanner 在信息缺口出现时增量修改计划(已完成任务强制保留,3 次熔断防死循环);
  • 危险操作(execute_python)走 HITL:AST 白名单 + 人工审批 + 否决终审。

架构

系统架构图

评测结果(100 条数据集,每模式 2 遍取均值)

指标 Fixed Adaptive
Task Success Rate 96% 97%
Tool Call Accuracy 见 评测报告
Plan Completion Rate 87% 92%
Re-plan Rate 0% 48%
Tokens / run 27,967 43,655
单遍成本(DeepSeek 空闲价) ¥6.0 ¥9.5

诚实结论:在易/中任务为主的通用调研任务分布上,两者成功率打平(逐题配对 3:3)—— Adaptive 的价值是保险而非免费午餐:收益集中在"首拆计划失效"的场景, 代价是 1.59 倍成本。完整分析(含失败归因与评估器校准)见 docs/experiments.md,评估器经 20 份盲测人工校准(一致率 85%)。

快速开始

# 1. 装依赖 + 准备钥匙
uv sync
cp .env.example .env    # 填入三把钥匙(DEEPSEEK_API_KEY / GITHUB_TOKEN / TAVILY_API_KEY)

# 2. 终端运行
uv run python main.py "分析 langchain-ai/langgraph 仓库并找中文教程"

# 交互会话(排队消息 / 两段式 Ctrl-C / 可编辑重发,见下节)
uv run python main.py

# 中断后恢复(Checkpoint 断点续跑)
uv run python main.py --resume <run_id>

交互模式:排队消息与两段式中断

不带目标启动即进入交互会话,三个交互承诺:

  1. 输入排队:agent 干活时打字不会被发出、也不会丢——进入队列,在下一个节点边界 投递为 user_update,由 replanner 并入计划(不打断也能转向);
  2. 两段式 Ctrl-C:第一次停在当前节点之后(SqliteSaver 已落盘快照,等价 --resume 的断点语义),2 秒窗口内第二次退出整个进程;
  3. 可编辑重发:软停后,排队消息预填进编辑框,改完回车经 Command(update=...) 注入白板;直接回车则从断点原样继续。

如实声明边界:进行中的 LLM/HTTP 调用无法中途取消(同步栈),软停在节点边界生效, 调用时长由 httpx 超时兜底;用户指示触发重规划同样受 3 次熔断保护。

实现一句话:prompt_toolkit 常驻输入行(patch_stdout 保证 worker 线程的进度打印 不打碎输入行)+ asyncio 主循环;图在 asyncio.to_thread 的 worker 线程里由 run_stream 逐节点消费,软中断旗(threading.Event)只在节点边界被检查。

跑评测(复现实验)

uv run python -m evals.run --mode adaptive     # 全量 100 条,约 1.5h
uv run python -m evals.judge --results-dir results/<目录>   # LLM-as-judge 阅卷
uv run python -m evals.calibrate make --n 20 --only-fails   # 人工盲测校准
uv run python -m evals.evaluate                # 生成对比报告

技术取舍

决策 选择 放弃 一句话理由
范式 Plan-and-Execute ReAct 全循环 控制流显式可调试,模型只管节点内决策
结构化输出 function calling 路线 json_schema 响应格式 实测兼容性更好(DeepSeek 思考模式不支持后者)
重试 指数退避,只重试瞬时故障 盲目重试 404/参数错重试一万次也没用
Checkpointer SQLite → Postgres 渐进 一开始就 PG 同一接口零逻辑替换,验证了抽象
记忆检索 BM25 + jieba 向量数据库 可解释、零基础设施;语义检索留给 RAG 项目
HITL interrupt + 人工否决终审 自动放行 人的否决必须是终审,否则审批是橡皮图章
交互层 prompt_toolkit + asyncio 会话 裸 input() + 线程 排队消息与预填重发是交互核心,值得一个依赖
评估 rubric + LLM-as-judge + 盲测校准 精确匹配 生成类任务没有唯一正确答案

完整决策记录(D1~D9)与风险预案见开发过程文档。

已知局限(如实声明)

  • judge 校验"要点命中"而不校验事实真伪,自洽的幻觉可能得分(已通过人工抽检缓解);
  • Retry 统计盲区:重试发生在包装器内,尚未回写状态字段;
  • eval 规模 100 条,±3pp 内差异不具统计显著性。

项目结构

见 docs/experiments.md(评测报告)、docs/devlog.md(六周开发实录,含全部踩坑记录)。

app/
├── agent/     # planner / executor / evaluator / replanner / human_gate / final_answer / graph / runner
├── tools/     # 注册表 + run_tool 包装器 + 7 个工具实现
├── models/    # Pydantic 数据模型(Task/Observation/EvaluationResult/Plan/AgentState)
├── ui/        # 交互会话(排队消息 / 两段式 Ctrl-C / 可编辑重发)
└── memory/    # save_memory + search_knowledge(BM25)
evals/         # 数据集 / runner / judge / 校准 / 指标
docs/          # devlog(六周实录)/ experiments.md(评测报告)/ traces

About

面向复杂技术任务的可动态重规划 Agent:任务拆解 → 工具执行 → 状态评估 → 动态重规划 → 断点恢复 → 人工审批。LangGraph + DeepSeek + FastAPI/SSE + PostgreSQL + Docker,附 100 条评测集 Fixed vs Adaptive 对照实验

Topics

Resources

Stars

0 stars

Watchers

0 watching

Forks

Releases

Packages

Contributors

Languages