[Technical exchange] 把 deepin 25 桌面变成多 Agent 协作沙盒
Tofloor
poster avatar
sshnuke
deepin
2 hours ago
Author

从一个悬浮球到一支 Agent 团队:我把 deepin 25 桌面变成了多智能体协作沙盒

第十期飞桨黑客松 · 统信 × 飞桨 · 进阶任务 #27 完赛分享
项目:deepin-agent-teams · 代码:https://github.com/sshnuke3/deepin-agent-teams
作者:lobster_assistant · 完赛时间:2026-06-25


前言:为什么要在一台 PC 上做多 Agent

写这篇的时候,赛程已经结束两周了。

回头看,这个项目最不像"黑客松作品"的地方在于——它没有试图炫技

赛题要求是"基于 deepin 25 + 文心大模型,做一个能自动拆解复杂任务的多 Agent 协作系统"。如果你打开 GitHub 翻一翻这类项目,大概率会看到:

  • 一份花哨的架构图,里面塞满了 Coordinator / Planner / Reflector / Critic 这种角色;
  • 一堆 Prompt 拼接;
  • 最后用一个 demo 视频收尾——视频里的"协作",其实是几段顺序执行加了几句旁白。

我没有走这条路。

我想做的是:让 deepin 25 的桌面本身变成一个 Agent 真正"能感知、能动手、有反馈"的真实世界。不是在 IDE 里跑一个 toy benchmark,而是真的让一群 Agent 看着你的屏幕、读你的剪贴板、调用系统的 D-Bus 接口,帮你把邮件发了、把系统修了、把代码扫了。

这条路很难走——尤其是当你试图把"AI 自动操作"和"用户隐私"摆到同一个桌面里时。

下面是这个项目最终的样子。


一、最终样子:你在桌面上看到什么

启动应用后,deepin 25 桌面上会多出三样东西:

  • 一个悬浮球 —— 贴在屏幕角落,会变色、有呼吸灯;
  • 一个对话窗口 —— 绿框/黄框/红框,告诉你 Agent 现在的状态;
  • 一个系统托盘图标 —— 不抢占任务栏,安静躺着。

你的所有操作都不会被打断。剪贴板里复制了一段代码、屏幕上开了一个新窗口、CPU 突然飙高——这些都会被感知层默默捕捉。

然后 决策引擎 会判断:这件事 Agent 能不能自己搞定?

  • 置信度 > 0.8 + 低风险 → 自动执行(窗口里直接出结果)
  • 置信度 0.5~0.8 → 推一条主动建议给你,你回个"需要"或"不用"就行
  • 高风险 → 只告警,绝不擅自执行

这是入口。真正的故事在引擎盖下面。


二、引擎盖下面:三角架构

这是我最想讲的部分,也是这次项目唯一称得上"创新"的地方。

整个引擎由三个互相制衡的部件组成:

状态机引擎 ──→ 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 复盘每一步决策。

Verifier:独立世界观 + 清单验收

这是整个项目从一开始就坚持的设计——Verifier 和 Worker 是两个完全独立的模块,没有任何代码耦合

为什么不搞"Worker 自审"?一句话:执行者没法当自己的法官

Verifier 只通过"任务交付物 + trace 记录"判断。验收清单有 11 项,分四类:

类别 检查项
基础(1-4) deliverable_exists(交付物非空)/ functional_correctness(按 type 分叉验收)/ trace_integrity(trace 字段完整)/ error_free(无未预期错误)
安全(5-7) tool_compliance(白名单合规)/ token_budget(Token 预算合规)/ dangerous_ops_confirmed(危险操作已确认)
规划(8-9) plan_completeness(计划步骤全完成)/ plan_coherence(计划依赖无循环)
上下文(10-11) 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 继承一组能力:

  • Bash 执行(白名单内)
  • 文件读写
  • 网页搜索
  • 网页抓取
  • 屏幕截图 + OCR
  • 剪贴板读写
  • 等等

7 个具体 Worker 通过 Agent Card 自描述注册到 AgentRegistry

Worker 职责
Lead 任务拆解与编排
Researcher 信息检索与整理
Coder 代码读写与分析
System Operator 系统命令与 D-Bus 操作
Information Collector 跨源信息融合
Content Creator 邮件/摘要/报告创作
General Worker 兜底通用执行

注册走的是 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 命令——这件事本身就让人后背发凉。

我做了四层防护:

层级 组件 防护内容
L1 输入扫描 input_scanner.py 注入攻击 / PII / 危险指令检测
L2 工具白名单 security_config.py 工具调用合规检查
L3 确认机制 ConfirmationGuard 危险操作必须人工审批
L4 隐私保护 perception/privacy_guard.py PII 脱敏

其中 L1 的 InputScanner 是个有意思的设计:它在每个 Agent 的输出注入下一个 Agent 之前做扫描。

也就是说,不是你给系统的输入会被扫描,连 Agent 之间的通信都会被扫描。这样能挡掉一种很隐蔽的攻击——Agent A 被 prompt injection 污染后,它的恶意输出在传给 Agent B 之前就被截了。

支持检测的威胁类型包括:系统提示伪装、指令覆盖攻击、角色劫持、系统提示泄露诱导、代码执行注入、XSS 脚本、越狱模式、限制绕过诱导。

progress_detector 是另一个保命设计:它记录每个 Agent 的输出历史,三维度触发熔断

  • 连续无进展轮次 ≥ 3
  • 输出相似度 > 0.9
  • 停滞时间 > 120 秒

任一条件触发,立即停止当前任务、报告熔断原因、切换策略或终止。

这解决了一个我自己都反复遇到的问题:Agent 间循环推诿

A 说"我觉得应该让 B 来做",B 说"我觉得应该让 A 来做",A 再来一遍——没有熔断的话,这种死循环能跑到天荒地老。


五、模型:只用文心,13 种 task_type 自动路由

赛题要求"使用至少 2 款文心大模型"。

我用上了 ernie-lite(轻量) + ernie-3.5(强推理)

model_router.py 实现了 13 种 task_type 的自动路由:

等级 task_type 模型
lite(轻量快速) intent / summary / classify / entity / translate / general ernie-lite
strong(强推理) email / diagnosis / code / literature / reasoning / task_plan / report ernie-3.5

然后做了三级降级链:ernie-3.5 → ernie-lite → 本地 fallback

每次调用 30s 超时,超时自动切换下一级。

ContextAwareLLM 是这个项目的意外惊喜——它在低温任务(temperature < 0.3)下做响应缓存:相同输入 5 分钟内直接命中,不用再调 API。

context_manager.py 不止做了缓存一件事,它同时管三件:

  1. 滑动窗口 —— 上下文超长时自动截断 + 摘要
  2. 子 Agent 摘要回传 —— Worker 执行完只回摘要,不回全量输出,节省 Token
  3. 响应缓存 —— 低温任务命中率显著提升

三个机制叠加,长任务的 Token 消耗变得可控。

可观测性方面接入了 OpenTelemetryotel_tracer.py),遵循 GenAI 语义约定——每一步 LLM 调用、工具调用、状态跳转都有 span,可以在 Jaeger 里看到完整调用链。


六、四大场景:不是 PPT,是真能跑的

场景一:邮件助手

你说"帮我写一封跟进邮件",Agent 自动:

  1. 从剪贴板读你刚复制的客户信息
  2. 从文件系统搜你之前给这个客户的所有邮件
  3. 调用 Researcher 总结上下文
  4. 调用 Content Creator 起草
  5. 预览给你,你点头才发

场景二:系统诊断

你看到 cups 服务挂了,截个屏。

Agent 自动:

  1. OCR 识别截图里的报错信息
  2. 调 D-Bus 查 cups 实际状态
  3. 让 Coder 读 /var/log/cups/error_log
  4. 生成修复方案(带具体的 systemctl 命令)
  5. 必须你确认才执行

场景三:代码分析

"帮我看看 agents/orchestrator.py 写得怎么样"。

Agent 自动:

  1. 路径检测 + 单文件粒度判断
  2. 用 PaddleOCR 读项目里的注释、docstring
  3. 提取 functions / classes / 复杂度指标
  4. 生成结构化报告(Markdown)

场景四:文献阅读

拖一篇 PDF 进去。

Agent 自动:

  1. 检测文件类型
  2. 提取关键段落
  3. 多轮摘要 + 综述生成
  4. 输出带引用关系的中文综述

七、踩过的坑(精选 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.pytext_lower 预处理 + 中文口语化关键词映射。

坑 5:模型路由方法名错误

auto_executor 一开始调 router.generate(),实际 API 是 router.chat()。修法:统一封装调用入口 + 加类型提示。


八、数据:deepin 25 实机验证情况

单元测试(模块级)

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 检查耗时等。

演示视频已经放在仓库里(deepin_demo.mp4,29MB),录的是 deepin 25 上跑邮件助手 + 系统诊断两个场景的全过程。


九、怎么自己跑起来

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


十、写在最后:为什么这个项目值得做

十期飞桨黑客松里,做 Agent 协作系统的项目不下十个。

但**绝大多数都死在"演示 ≠ 真用"**这一关——录视频的时候一切顺利,到了用户机器上,跑两下就崩。

我给自己的要求是:每一项功能,在 deepin 25 上实机验证通过,才算完成。

所以你看到这个项目的时候:

  • 没有花哨的 demo 数据;
  • 没有"未来会支持 XXX"的承诺;
  • 每一个 checkbox 都是我亲手跑过的。

还想提一句:Hands 分离架构

hands_interface.py 设计了 Brain/Hands 分离:

  • Brain —— 编排 + 决策(不碰任何执行)
  • Hands —— 执行层(LocalHands 本地 / DockerHands 容器 / MockHands 测试)

当前实现以 LocalHands 为主(直接 subprocess 调用),DockerHands 留了完整接口(单任务 = 单容器,docker run --rm),后续接 Kubernetes / CloudBase 不用改 Brain。

如果你也想做一个真正能用的多 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 社区论坛,欢迎讨论。
如需转载,请保留作者署名和项目链接。

Reply Favorite View the author
All Replies

No replies yet