第十期飞桨黑客松 · 统信 × 飞桨 · 进阶任务 #27 完赛分享 项目:deepin-agent-teams · 代码:https://github.com/sshnuke3/deepin-agent-teams 作者:lobster_assistant · 完赛时间:2026-06-25
写这篇的时候,赛程已经结束两周了。
回头看,这个项目最不像"黑客松作品"的地方在于——它没有试图炫技。
赛题要求是"基于 deepin 25 + 文心大模型,做一个能自动拆解复杂任务的多 Agent 协作系统"。如果你打开 GitHub 翻一翻这类项目,大概率会看到:
我没有走这条路。
我想做的是:让 deepin 25 的桌面本身变成一个 Agent 真正"能感知、能动手、有反馈"的真实世界。不是在 IDE 里跑一个 toy benchmark,而是真的让一群 Agent 看着你的屏幕、读你的剪贴板、调用系统的 D-Bus 接口,帮你把邮件发了、把系统修了、把代码扫了。
这条路很难走——尤其是当你试图把"AI 自动操作"和"用户隐私"摆到同一个桌面里时。
下面是这个项目最终的样子。
启动应用后,deepin 25 桌面上会多出三样东西:
你的所有操作都不会被打断。剪贴板里复制了一段代码、屏幕上开了一个新窗口、CPU 突然飙高——这些都会被感知层默默捕捉。
然后 决策引擎 会判断:这件事 Agent 能不能自己搞定?
这是入口。真正的故事在引擎盖下面。
这是我最想讲的部分,也是这次项目唯一称得上"创新"的地方。
整个引擎由三个互相制衡的部件组成:
状态机引擎 ──→ Worker 池(执行者)
↑
│
Verifier(独立质检)
PENDING → CLAIMED → PLANNING → RUNNING → VERIFIED → COMPLETED
↓ FAIL
RETRY(≤3次)
↓ 超次
FAILED
注意:这些跳转条件不是模型判断的,是 if/else 写死的。
为什么?因为我被 LLM 自己的"主观能动性"坑过太多次。Agent 说"我觉得这件事可以结束了"——可它判错了怎么办?
把跳转条件写死之后,整个流程有了确定性。你可以在脑子里画一条线,从开始到结束只有 7 个状态,每个状态进入的条件和退出的条件都是明确的。
代价是什么?代价是所有"模糊判断"都被推给了其他组件(比如 Verifier,比如 Worker 自己)。
每次跳转写一行 JSONL 到 data/traces/{task_id}.jsonl,可以拿 tools/analyze_traces.py 复盘每一步决策。
data/traces/{task_id}.jsonl
tools/analyze_traces.py
这是整个项目从一开始就坚持的设计——Verifier 和 Worker 是两个完全独立的模块,没有任何代码耦合。
为什么不搞"Worker 自审"?一句话:执行者没法当自己的法官。
Verifier 只通过"任务交付物 + trace 记录"判断。验收清单有 11 项,分四类:
deliverable_exists
functional_correctness
trace_integrity
error_free
tool_compliance
token_budget
dangerous_ops_confirmed
plan_completeness
plan_coherence
context_overflow
summary_quality
每项独立打分,最终结果是 PASS / FAIL(causes[]) / RETRY(cause) 三选一。
有意思的是:Verifier 不仅在三角架构里做质检,还是辩论模式(debate.py)的 Judge。当 Pro Agent 和 Con Agent 争论时,最终裁决交给 Verifier 做——同一套验收标准,两种场景复用。
Worker 是真正的执行者。它知道自己能做什么、不能做什么,能用什么工具、不能用哪些工具。
每个 Worker 启动时都会从 worker_base.py 继承一组能力:
worker_base.py
7 个具体 Worker 通过 Agent Card 自描述注册到 AgentRegistry:
AgentRegistry
注册走的是 Google 提出的 A2A(Agent-to-Agent)协议风格——每个 Worker 一个 JSON 描述文件,registry 自动发现 + 心跳维护。这意味着以后加新 Worker = 丢一张 JSON 进 agents/agent_cards/,不用改引擎代码。
agents/agent_cards/
这套设计带来的好处是:添加新场景 = 写一个新 Worker,不用改引擎本身。
项目里有 3 个 MCP Server:
model_server.py
file_server.py
system_server.py
Agent 想要调工具,不再直接 import 函数,而是通过 JSON-RPC over stdio 发请求给 MCP Server。
这意味着什么?意味着我可以在不重启 Agent 的前提下替换任何一个工具的实现。
比如我想把"文件搜索"从本地 find 换成 ripgrep——改 MCP Server 一行代码就行,Agent 那边 0 改动。
find
ripgrep
更深一层的好处:以后接 Notion、接 Slack、接飞书,只要写一个新的 MCP Server 注册到 registry 里就行,主流程代码完全不动。
顺便提一句:MCP 协议不是调第三方 SDK,是我自己用 200 行 Python 实现的 JSON-RPC over stdio。理由很简单——少一个外部依赖,主流程就少一处可能出问题的环节。
这个项目最容易翻车的地方不是技术,是安全。
Agent 真的能读你的剪贴板、截你的屏、跑 shell 命令——这件事本身就让人后背发凉。
我做了四层防护:
input_scanner.py
security_config.py
ConfirmationGuard
perception/privacy_guard.py
其中 L1 的 InputScanner 是个有意思的设计:它在每个 Agent 的输出注入下一个 Agent 之前做扫描。
也就是说,不是你给系统的输入会被扫描,连 Agent 之间的通信都会被扫描。这样能挡掉一种很隐蔽的攻击——Agent A 被 prompt injection 污染后,它的恶意输出在传给 Agent B 之前就被截了。
支持检测的威胁类型包括:系统提示伪装、指令覆盖攻击、角色劫持、系统提示泄露诱导、代码执行注入、XSS 脚本、越狱模式、限制绕过诱导。
progress_detector 是另一个保命设计:它记录每个 Agent 的输出历史,三维度触发熔断:
progress_detector
任一条件触发,立即停止当前任务、报告熔断原因、切换策略或终止。
这解决了一个我自己都反复遇到的问题:Agent 间循环推诿。
A 说"我觉得应该让 B 来做",B 说"我觉得应该让 A 来做",A 再来一遍——没有熔断的话,这种死循环能跑到天荒地老。
赛题要求"使用至少 2 款文心大模型"。
我用上了 ernie-lite(轻量) + ernie-3.5(强推理)。
model_router.py 实现了 13 种 task_type 的自动路由:
model_router.py
然后做了三级降级链:ernie-3.5 → ernie-lite → 本地 fallback。
每次调用 30s 超时,超时自动切换下一级。
ContextAwareLLM 是这个项目的意外惊喜——它在低温任务(temperature < 0.3)下做响应缓存:相同输入 5 分钟内直接命中,不用再调 API。
ContextAwareLLM
context_manager.py 不止做了缓存一件事,它同时管三件:
context_manager.py
三个机制叠加,长任务的 Token 消耗变得可控。
可观测性方面接入了 OpenTelemetry(otel_tracer.py),遵循 GenAI 语义约定——每一步 LLM 调用、工具调用、状态跳转都有 span,可以在 Jaeger 里看到完整调用链。
otel_tracer.py
你说"帮我写一封跟进邮件",Agent 自动:
你看到 cups 服务挂了,截个屏。
Agent 自动:
/var/log/cups/error_log
"帮我看看 agents/orchestrator.py 写得怎么样"。
agents/orchestrator.py
拖一篇 PDF 进去。
A 让 B 做,B 让 A 做。修法:progress_detector 三维度熔断(见第四章)。
一个长任务跑到一半,Worker 把整个 trace 全塞回上下文,窗口爆了。修法:context_manager.py 滑动窗口 + 子 Agent 摘要回传 + 缓存三件套。
GUI 在 deepin 25 上启动时,PyQt5 的 FramelessWindowHint 跟 fcitx 打架。修法:去掉这个 flag,加 QT_QPA_PLATFORM=xcb。
FramelessWindowHint
QT_QPA_PLATFORM=xcb
用户回"需要/要/行/ok"——分类器只认 yes/no。修法:scenario_classifier.py 加 text_lower 预处理 + 中文口语化关键词映射。
scenario_classifier.py
text_lower
auto_executor 一开始调 router.generate(),实际 API 是 router.chat()。修法:统一封装调用入口 + 加类型提示。
auto_executor
router.generate()
router.chat()
task_state_machine 5/5 ✅ verifier 6/6 ✅ checkpoint_manager 6/6 ✅ model_router 4/4 ✅ tool_registry 8/8 ✅ mcp_integration 8/8 ✅ ───────────────────────────────── 小计 34/34 ✅
test_e2e.py 11/11 ✅ - Orchestrator 初始化(tools / workers 两种模式) - LLM 健康预检 - 任务分解 + 本地工具执行 - Verifier 验收 - 完整 pipeline - 超时与重试 - 部分失败优雅降级 - 状态机生命周期 - Verifier 基础能力 - Orchestrator 工厂
tests/benchmark.py 跑了 8 个模块的性能基准,包括 LLM 调用吞吐、MCP 通信延迟、状态机跳转开销、Verifier 检查耗时等。
tests/benchmark.py
演示视频已经放在仓库里(deepin_demo.mp4,29MB),录的是 deepin 25 上跑邮件助手 + 系统诊断两个场景的全过程。
deepin_demo.mp4
git clone https://github.com/sshnuke3/deepin-agent-teams.git cd deepin-agent-teams # 装依赖 pip install -r requirements.txt # 配文心 API key(写到 .env) echo "ERNIE_API_KEY=***" > .env # 一键部署(deepin 25) bash deepin25_deploy.sh # 启动 GUI python3 main.py --gui # 或者直接跑编排器(推荐 v4) python3 agents/orchestrator_v4.py
详细的部署文档在 DEPLOYMENT.md,技术架构在 docs/ARCHITECTURE.md,技术决策记录在 docs/TECH_DECISIONS.md。
DEPLOYMENT.md
docs/ARCHITECTURE.md
docs/TECH_DECISIONS.md
十期飞桨黑客松里,做 Agent 协作系统的项目不下十个。
但**绝大多数都死在"演示 ≠ 真用"**这一关——录视频的时候一切顺利,到了用户机器上,跑两下就崩。
我给自己的要求是:每一项功能,在 deepin 25 上实机验证通过,才算完成。
所以你看到这个项目的时候:
hands_interface.py 设计了 Brain/Hands 分离:
hands_interface.py
当前实现以 LocalHands 为主(直接 subprocess 调用),DockerHands 留了完整接口(单任务 = 单容器,docker run --rm),后续接 Kubernetes / CloudBase 不用改 Brain。
docker run --rm
如果你也想做一个真正能用的多 Agent 系统,希望这个项目能给你一些参考。
代码在 GitHub 上,欢迎提 issue 和 PR:
🔗 https://github.com/sshnuke3/deepin-agent-teams
附:项目结构速览
deepin-agent-teams/ ├── agents/ # 编排器 + 状态机 + Verifier + Worker 基类 │ ├── orchestrator.py # 统一编排器(fan_out 并行 + aggregate 聚合) │ ├── orchestrator_v3.py # v3 编排器(原版) │ ├── orchestrator_v4.py # v4 编排器(MCP 驱动) │ ├── task_state_machine.py # 7 状态状态机引擎(含 PLANNING) │ ├── verifier.py # 独立质检员(11 项检查) │ ├── debate.py # 辩论模式(Pro / Con / Judge) │ ├── hands_interface.py # Brain / Hands 分离架构 │ ├── otel_tracer.py # OpenTelemetry 封装(GenAI 语义约定) │ ├── model_router.py # 文心模型路由(13 种 task_type) │ ├── context_manager.py # 上下文三件套(窗口 + 摘要 + 缓存) │ ├── input_scanner.py # Agent 间通信安全扫描 │ ├── progress_detector.py # 三维度熔断 │ └── agent_cards/ # 7 张 Agent Card(A2A 风格) ├── mcp_servers/ # 3 个 MCP Server + 200 行协议实现 ├── perception/ # 屏幕/OCR/剪贴板/D-Bus 感知层 ├── scenarios/ # 4 大场景(邮件/诊断/代码/文献) ├── gui/ # PyQt5 悬浮球 + 对话窗口 + 决策引擎 ├── prompts/ # 10 个热加载 Prompt 模板 ├── data/ │ ├── traces/ # JSONL 任务追踪 │ └── checkpoints/ # SQLite 检查点 ├── docs/ # ARCHITECTURE / TECH_DECISIONS / QUALITY └── tests/ # 11 个 e2e + 8 模块 benchmark
本文首发于 deepin 社区论坛,欢迎讨论。 如需转载,请保留作者署名和项目链接。
No replies yet
Featured Collection
Popular Ranking
Popular Events
从一个悬浮球到一支 Agent 团队:我把 deepin 25 桌面变成了多智能体协作沙盒
前言:为什么要在一台 PC 上做多 Agent
写这篇的时候,赛程已经结束两周了。
回头看,这个项目最不像"黑客松作品"的地方在于——它没有试图炫技。
赛题要求是"基于 deepin 25 + 文心大模型,做一个能自动拆解复杂任务的多 Agent 协作系统"。如果你打开 GitHub 翻一翻这类项目,大概率会看到:
我没有走这条路。
我想做的是:让 deepin 25 的桌面本身变成一个 Agent 真正"能感知、能动手、有反馈"的真实世界。不是在 IDE 里跑一个 toy benchmark,而是真的让一群 Agent 看着你的屏幕、读你的剪贴板、调用系统的 D-Bus 接口,帮你把邮件发了、把系统修了、把代码扫了。
这条路很难走——尤其是当你试图把"AI 自动操作"和"用户隐私"摆到同一个桌面里时。
下面是这个项目最终的样子。
一、最终样子:你在桌面上看到什么
启动应用后,deepin 25 桌面上会多出三样东西:
你的所有操作都不会被打断。剪贴板里复制了一段代码、屏幕上开了一个新窗口、CPU 突然飙高——这些都会被感知层默默捕捉。
然后 决策引擎 会判断:这件事 Agent 能不能自己搞定?
这是入口。真正的故事在引擎盖下面。
二、引擎盖下面:三角架构
这是我最想讲的部分,也是这次项目唯一称得上"创新"的地方。
整个引擎由三个互相制衡的部件组成:
状态机引擎 ──→ Worker 池(执行者)
↑
│
Verifier(独立质检)
状态机引擎:所有停止条件写死在代码里
↓ FAIL
RETRY(≤3次)
↓ 超次
FAILED
注意:这些跳转条件不是模型判断的,是 if/else 写死的。
为什么?因为我被 LLM 自己的"主观能动性"坑过太多次。Agent 说"我觉得这件事可以结束了"——可它判错了怎么办?
把跳转条件写死之后,整个流程有了确定性。你可以在脑子里画一条线,从开始到结束只有 7 个状态,每个状态进入的条件和退出的条件都是明确的。
代价是什么?代价是所有"模糊判断"都被推给了其他组件(比如 Verifier,比如 Worker 自己)。
每次跳转写一行 JSONL 到
data/traces/{task_id}.jsonl,可以拿tools/analyze_traces.py复盘每一步决策。Verifier:独立世界观 + 清单验收
这是整个项目从一开始就坚持的设计——Verifier 和 Worker 是两个完全独立的模块,没有任何代码耦合。
为什么不搞"Worker 自审"?一句话:执行者没法当自己的法官。
Verifier 只通过"任务交付物 + trace 记录"判断。验收清单有 11 项,分四类:
deliverable_exists(交付物非空)/functional_correctness(按 type 分叉验收)/trace_integrity(trace 字段完整)/error_free(无未预期错误)tool_compliance(白名单合规)/token_budget(Token 预算合规)/dangerous_ops_confirmed(危险操作已确认)plan_completeness(计划步骤全完成)/plan_coherence(计划依赖无循环)context_overflow(上下文不溢出)/summary_quality(子 Agent 摘要保留关键信息)每项独立打分,最终结果是 PASS / FAIL(causes[]) / RETRY(cause) 三选一。
有意思的是:Verifier 不仅在三角架构里做质检,还是辩论模式(debate.py)的 Judge。当 Pro Agent 和 Con Agent 争论时,最终裁决交给 Verifier 做——同一套验收标准,两种场景复用。
Worker 池:14 种能力,各司其职
Worker 是真正的执行者。它知道自己能做什么、不能做什么,能用什么工具、不能用哪些工具。
每个 Worker 启动时都会从
worker_base.py继承一组能力:7 个具体 Worker 通过 Agent Card 自描述注册到
AgentRegistry:注册走的是 Google 提出的 A2A(Agent-to-Agent)协议风格——每个 Worker 一个 JSON 描述文件,registry 自动发现 + 心跳维护。这意味着以后加新 Worker = 丢一张 JSON 进
agents/agent_cards/,不用改引擎代码。这套设计带来的好处是:添加新场景 = 写一个新 Worker,不用改引擎本身。
三、MCP:把工具从 Agent 里拆出来
项目里有 3 个 MCP Server:
model_server.py—— 模型调用(统一封装文心 API)file_server.py—— 文件操作(带权限校验)system_server.py—— 系统操作(带 D-Bus 桥接)Agent 想要调工具,不再直接 import 函数,而是通过 JSON-RPC over stdio 发请求给 MCP Server。
这意味着什么?意味着我可以在不重启 Agent 的前提下替换任何一个工具的实现。
比如我想把"文件搜索"从本地
find换成ripgrep——改 MCP Server 一行代码就行,Agent 那边 0 改动。更深一层的好处:以后接 Notion、接 Slack、接飞书,只要写一个新的 MCP Server 注册到 registry 里就行,主流程代码完全不动。
顺便提一句:MCP 协议不是调第三方 SDK,是我自己用 200 行 Python 实现的 JSON-RPC over stdio。理由很简单——少一个外部依赖,主流程就少一处可能出问题的环节。
四、安全:把"AI 操作"和"用户隐私"摆到同一个桌面
这个项目最容易翻车的地方不是技术,是安全。
Agent 真的能读你的剪贴板、截你的屏、跑 shell 命令——这件事本身就让人后背发凉。
我做了四层防护:
input_scanner.pysecurity_config.pyConfirmationGuardperception/privacy_guard.py其中 L1 的 InputScanner 是个有意思的设计:它在每个 Agent 的输出注入下一个 Agent 之前做扫描。
也就是说,不是你给系统的输入会被扫描,连 Agent 之间的通信都会被扫描。这样能挡掉一种很隐蔽的攻击——Agent A 被 prompt injection 污染后,它的恶意输出在传给 Agent B 之前就被截了。
支持检测的威胁类型包括:系统提示伪装、指令覆盖攻击、角色劫持、系统提示泄露诱导、代码执行注入、XSS 脚本、越狱模式、限制绕过诱导。
progress_detector是另一个保命设计:它记录每个 Agent 的输出历史,三维度触发熔断:任一条件触发,立即停止当前任务、报告熔断原因、切换策略或终止。
这解决了一个我自己都反复遇到的问题:Agent 间循环推诿。
A 说"我觉得应该让 B 来做",B 说"我觉得应该让 A 来做",A 再来一遍——没有熔断的话,这种死循环能跑到天荒地老。
五、模型:只用文心,13 种 task_type 自动路由
赛题要求"使用至少 2 款文心大模型"。
我用上了 ernie-lite(轻量) + ernie-3.5(强推理)。
model_router.py实现了 13 种 task_type 的自动路由:然后做了三级降级链:ernie-3.5 → ernie-lite → 本地 fallback。
每次调用 30s 超时,超时自动切换下一级。
ContextAwareLLM是这个项目的意外惊喜——它在低温任务(temperature < 0.3)下做响应缓存:相同输入 5 分钟内直接命中,不用再调 API。context_manager.py不止做了缓存一件事,它同时管三件:三个机制叠加,长任务的 Token 消耗变得可控。
可观测性方面接入了 OpenTelemetry(
otel_tracer.py),遵循 GenAI 语义约定——每一步 LLM 调用、工具调用、状态跳转都有 span,可以在 Jaeger 里看到完整调用链。六、四大场景:不是 PPT,是真能跑的
场景一:邮件助手
你说"帮我写一封跟进邮件",Agent 自动:
场景二:系统诊断
你看到 cups 服务挂了,截个屏。
Agent 自动:
/var/log/cups/error_log场景三:代码分析
"帮我看看
agents/orchestrator.py写得怎么样"。Agent 自动:
场景四:文献阅读
拖一篇 PDF 进去。
Agent 自动:
七、踩过的坑(精选 5 个)
坑 1:循环推诿
A 让 B 做,B 让 A 做。修法:
progress_detector三维度熔断(见第四章)。坑 2:Token 预算爆掉
一个长任务跑到一半,Worker 把整个 trace 全塞回上下文,窗口爆了。修法:
context_manager.py滑动窗口 + 子 Agent 摘要回传 + 缓存三件套。坑 3:deepin 输入法冲突
GUI 在 deepin 25 上启动时,PyQt5 的
FramelessWindowHint跟 fcitx 打架。修法:去掉这个 flag,加QT_QPA_PLATFORM=xcb。坑 4:场景分类器对口语化确认词不识别
用户回"需要/要/行/ok"——分类器只认 yes/no。修法:
scenario_classifier.py加text_lower预处理 + 中文口语化关键词映射。坑 5:模型路由方法名错误
auto_executor一开始调router.generate(),实际 API 是router.chat()。修法:统一封装调用入口 + 加类型提示。八、数据:deepin 25 实机验证情况
单元测试(模块级)
端到端测试(集成级)
性能基准
tests/benchmark.py跑了 8 个模块的性能基准,包括 LLM 调用吞吐、MCP 通信延迟、状态机跳转开销、Verifier 检查耗时等。演示视频已经放在仓库里(
deepin_demo.mp4,29MB),录的是 deepin 25 上跑邮件助手 + 系统诊断两个场景的全过程。九、怎么自己跑起来
详细的部署文档在
DEPLOYMENT.md,技术架构在docs/ARCHITECTURE.md,技术决策记录在docs/TECH_DECISIONS.md。十、写在最后:为什么这个项目值得做
十期飞桨黑客松里,做 Agent 协作系统的项目不下十个。
但**绝大多数都死在"演示 ≠ 真用"**这一关——录视频的时候一切顺利,到了用户机器上,跑两下就崩。
我给自己的要求是:每一项功能,在 deepin 25 上实机验证通过,才算完成。
所以你看到这个项目的时候:
还想提一句:Hands 分离架构
hands_interface.py设计了 Brain/Hands 分离:当前实现以 LocalHands 为主(直接 subprocess 调用),DockerHands 留了完整接口(单任务 = 单容器,
docker run --rm),后续接 Kubernetes / CloudBase 不用改 Brain。如果你也想做一个真正能用的多 Agent 系统,希望这个项目能给你一些参考。
代码在 GitHub 上,欢迎提 issue 和 PR:
🔗 https://github.com/sshnuke3/deepin-agent-teams
附:项目结构速览