把一个想法做成可运行的多 Agent 协作应用：一句话进，一组 Agent 按可配置的协作拓扑分工完成，全过程实时可视、产物自动归档、轨迹可回溯。
重点不在"用了几个 Agent"，而在编排内核——它把多 Agent 协作真正难的四件事收成一层：拓扑调度、共享状态、流式可观测、失败降级。场景只是挂在它上面的配置。
# 1) 装依赖（只有 4 个包）
python -m venv .venv
source .venv/Scripts/activate # Windows Git Bash；PowerShell 用 .venv\Scripts\Activate.ps1；macOS/Linux 用 source .venv/bin/activate
pip install -r requirements.txt
# 2) 启动（不需要任何配置）
python run.py打开 http://127.0.0.1:8000/，选一个场景、填一句话任务、点「开始协作」。
零配置就能看到完整效果：没有配置
LLM_API_KEY时自动进入 Mock 模式，三个场景都会用贴合任务的内置示例内容把全流程跑完——包括工具调用、DAG 并发、评审回流。想接真实模型，复制.env.example为.env填上 Key 即可。
接真实模型（OpenAI 兼容协议，DeepSeek / 通义 / Kimi / 智谱 / vLLM / Ollama 均可）：
cp .env.example .env
# 编辑 .env：
# LLM_API_KEY=sk-xxxx
# LLM_BASE_URL=https://api.deepseek.com/v1
# LLM_MODEL_STRONG=deepseek-reasoner
python run.py| 场景 | 协作拓扑 | 角色 | 这个场景用来验证什么 |
|---|---|---|---|
| 多 Agent 研发小组 | 主管编排式 | 主管 · 架构师 · 后端 · 前端 · 测试 · 文档 | 结构化输出驱动控制流：主管输出的 JSON 直接变成任务 DAG；无依赖的子任务真并发 |
| 多 Agent 深度研究助手 | 顺序流水线 | 拆题 · 检索 · 核查 · 撰写 · 评审 | 条件回流：评审判定「需修改」会自动回到撰写重做再复评；Agent 自发调用工具并回灌结果 |
| 多 Agent 辩论决策团 | 群聊辩论式 | 支持方 · 质询方 · 主持人 · 裁判 | 多轮循环 + 提前退出：主持人判断分歧已收敛就结束，不把 token 烧在重复论点上 |
三种拓扑共用一个调度器、一块共享黑板、一套事件流。换场景内核代码零改动。
浏览器控制台（零构建，原生 HTML/CSS/JS）
│ POST /api/runs ──▶ 创建任务，立即返回 run_id
│ GET /api/runs/{id}/events ──▶ SSE 实时事件流
▼
FastAPI 接口层 统一响应体 · 参数校验 · 全局异常
▼
编排内核 Orchestrator ← 这层是项目的核心
├─ 拓扑调度：顺序流水线 / 主管编排(DAG并发) / 群聊辩论(可提前收敛)
├─ 共享黑板：消息流 · 产物池 · 任务 DAG · 事实与决策
├─ 工具回路：指令式调用 → 内核执行 → 结果回灌 → 重新综合
└─ 失败处理：重试 → 降级为 Mock 继续跑完 → degraded 标记带到前端
▼
LLM 网关 多模型路由(fast/default/strong) · OpenAI 兼容 · 自动降级
▼ SQLite（schema.sql 为标准 SQL，可迁 MySQL/PostgreSQL）
目录结构：
multiagent-studio/
├── run.py # 启动入口
├── schema.sql # 数据库结构（含 MySQL 迁移说明）
├── requirements.txt
├── .env.example
├── app/
│ ├── main.py # 应用装配、静态托管、全局异常
│ ├── config.py # 全部配置走环境变量
│ ├── db.py # 参数化 SQL 数据访问层
│ ├── services.py # 运行时装配（网关/工具/场景/内核）
│ ├── core/
│ │ ├── engine.py # 编排内核：三种拓扑 + 事件总线 + 黑板调度
│ │ ├── llm.py # LLM 网关：流式 + 重试 + 降级
│ │ ├── tools.py # 工具注册中心（MCP 风格，危险工具默认关闭）
│ │ ├── scenario.py # 场景/角色的声明式模型
│ │ └── message.py # 消息、产物、任务节点、共享黑板
│ ├── api/
│ │ ├── routes.py # HTTP 接口 + SSE
│ │ └── schemas.py # 入参校验模型
│ └── scenarios/ # 场景插件（加场景只动这里）
│ ├── dev_squad.py
│ ├── research.py
│ └── debate.py
├── web/ # 控制台（index.html / app.js / style.css）
├── scripts/
│ ├── smoke_test.py # 内核冒烟测试（不开 HTTP）
│ └── api_test.py # 接口联调测试（需要服务已启动）
└── deploy/nginx.conf # 含 SSE 必需的反代配置
| 表 | 作用 |
|---|---|
runs |
一次运行：场景、任务、状态、拓扑、token 用量、步数、是否降级 |
messages |
完整协作轨迹：谁在第几步说了什么、用的哪个模型、耗时多少 token |
artifacts |
产物正文：PRD、架构设计、代码、测试报告、裁决报告 |
tasks |
主管编排拓扑的任务 DAG：负责人、依赖、状态、产出 |
轨迹全量落库的意义：回放不重跑模型。点历史运行即可重新渲染整条协作过程，不产生任何 API 费用——这既是调试手段，也是评测的数据基础。
| 方法 | 路径 | 说明 |
|---|---|---|
| GET | /api/health |
健康检查 + 运行模式 + 已加载场景与工具 |
| GET | /api/scenarios |
场景清单（含拓扑与角色分工） |
| POST | /api/runs |
创建任务 {scenario, task}，立即返回 run_id |
| GET | /api/runs |
历史运行列表 |
| GET | /api/runs/{id}/events |
SSE 实时事件流（先补播历史再持续推送，刷新不丢进度） |
| GET | /api/runs/{id} |
轨迹 + 产物列表 + 任务 DAG |
| DELETE | /api/runs/{id} |
删除运行记录 |
| GET | /api/artifacts/{id} |
产物正文 |
| GET | /api/artifacts/{id}/download |
下载产物为 Markdown |
交互式 API 文档：http://127.0.0.1:8000/docs
事件类型：run_started · agent_started · delta（流式分片）· agent_finished · message（工具消息）· task_update · artifact · notice · error · run_finished
内核不认识任何业务，加场景是加配置而不是改引擎。三步：
第一步，在 app/scenarios/ 下新建模块，用 AgentSpec 描述角色：
from ..core.scenario import AgentSpec, Scenario, TOPOLOGY_SUPERVISOR
AGENTS = [
AgentSpec(
key="leader", name="组长", role="拆解与汇总",
system="你是组长……", # 角色系统提示
tier="strong", # 走强模型档
intent="把需求拆成带依赖的子任务",
mock=lambda task, prompt, bb: "（无 Key 时用的示例输出）",
),
# …其余角色
]
def build() -> Scenario:
return Scenario(
key="my_scene", name="我的场景", description="……",
topology=TOPOLOGY_SUPERVISOR, # 复用已有拓扑
agents=AGENTS,
config={"lead": "leader", "fallback_plan": [...], "final_artifact": {...}},
)第二步，在 app/scenarios/__init__.py 的 SCENARIO_MODULES 里加一行。
第三步，重启。场景自动出现在控制台里。
想要一种新拓扑（比如黑板共享式、投票式），只需在 app/core/engine.py 里加一个 _run_xxx 方法并注册进 _DISPATCH，所有场景都能用。
# 内核冒烟测试：不开 HTTP，直接驱动引擎跑通三个场景
python scripts/smoke_test.py
# 接口联调测试：需要先 python run.py
python scripts/api_test.py冒烟测试覆盖：消息与产物落库一致性、用量统计一致性、DAG 子任务全部完成、 工具调用真实发生、评审回流生效（首评需修改 → 复评通过）、辩论轮次与裁决产出。
接口测试覆盖：SSE 流式分片数量与响应头、禁缓冲头、参数校验错误码、 轨迹与产物查询、下载头编码、回放不重复写入、清理。
cp .env.example .env # 填好 LLM_API_KEY
docker compose up -d --build
# 访问 http://127.0.0.1:8080pip install -r requirements.txt
HOST=0.0.0.0 PORT=8000 python run.py # 建议交给 systemd 托管deploy/nginx.conf 已包含 SSE 必需的三个配置（proxy_buffering off、
proxy_cache off、proxy_read_timeout 拉长）。少了这几行，流式响应会变成
「卡很久然后一次性全吐出来」——这是流式应用上线最常见的坑。
-
.env不进仓库（已在.gitignore里） - 生产环境
RELOAD=0 - 数据库文件目录挂到持久卷（
studio-data） -
ALLOW_CODE_EXEC保持false，除非确认要执行模型产出的代码 - Nginx 配 HTTPS，SSE 长连接记得放宽
proxy_read_timeout - 回滚方案：镜像 tag 用 commit 短哈希，禁止
latest；SQLite 库定期备份
- 任务在单进程内存里跑：重启服务会中断进行中的运行（已完成轨迹已落库）。 多副本部署需要把运行态搬到 Redis 或改为任务队列消费。
- 事件总线是进程内的：多实例下 SSE 只能连到跑该任务的那个实例， 需要引入 Redis Pub/Sub 做跨实例广播。
- 工具调用走指令式解析而非模型原生 function calling：兼容性最好，
但不如原生调用稳定。换成原生
tools参数是明确的升级路径。 - Mock 模式的输出是内置示例：用来验证流程与演示，不代表真实模型能力。
本项目按「内核稳定 → 场景增量」推进。以下按优先级排列，每完成一项会在 CHANGELOG.md 里留下对应记录。
近期（内核短板）
- 运行态外置：把任务状态与事件总线搬到 Redis，支持多副本部署与断点续跑
- 跨实例 SSE 广播：引入 Redis Pub/Sub，解决多实例下事件只能连到本机的问题
- 原生 function calling：把指令式工具解析换成模型原生
tools参数，提升调用稳定性 - 重试粒度细化：目前重试只在连接阶段，需支持单个子任务失败后局部重试而不整体重跑
中期（能力扩展）
- 新拓扑：黑板共享式：多个 Agent 轮询认领任务，适合无固定分工的开放型任务
- 新拓扑：投票式：同角色多实例并行产出后投票裁决，用于降低单点幻觉
- 人工介入节点：在关键步骤插入人工确认（human-in-the-loop），支持暂停与恢复
- 成本与质量看板：按场景/角色聚合 token 成本、成功率、平均步数，暴露无效循环
长期（工程纵深）
- 评测集与回归：为三个场景建立固定评测集，用轨迹级指标（步数、无效调用比例）做回归
- 前端重构为 Vue 3 + Vite：接口层已完全分离，属于纯替换，不影响后端
- 可观测接入 OpenTelemetry：把每次协作的 span 导出，接 Jaeger / Tempo
想贡献某个方向？开 issue 说明打算做哪一项，避免多人重复劳动。
性能
- DAG 无依赖子任务并发执行，并发度由
MAX_CONCURRENCY控制（默认 2，防上游限流） - 共享黑板对上下文做摘要压缩（
transcript/artifacts_digest都有字数上限）， 避免长链路下上下文线性膨胀 messages按(run_id, seq)建索引，runs按created_at DESC建索引
安全
- 所有 SQL 使用占位符参数化，无字符串拼接
- 密钥全部走环境变量，仓库内零密钥；日志不打印任何密钥
- 高危工具（真实代码执行）默认关闭，需显式开
ALLOW_CODE_EXEC - 模型输出以
textContent与转义后 HTML 渲染，不直接注入 DOM - 内部异常不外泄堆栈，只返回统一错误码，细节留在服务端日志