multiAgentTry/docs/backend-steps.md

后端开发步骤

每一步完成后都有验证方法，前端逐步替换 mock 数据。

---

第一步：项目初始化

操作

cd backend
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate
pip install fastapi uvicorn pydantic typer aiofiles
mkdir -p .doc/agents .doc/cache .doc/meetings .doc/resources .doc/workflow

创建文件

• backend/main.py - FastAPI 入口
• backend/cli.py - CLI 入口

验证

python cli.py hello  # 输出: Hello Swarm
python -m uvicorn main:app --reload
curl http://localhost:8000/  # 输出: {"status":"ok"}

---

第二步：基础存储服务

操作 - 创建 backend/services/storage.py

class StorageService:
    async def read_json(self, path: str) -> dict
    async def write_json(self, path: str, data: dict)
    async def ensure_dir(self, path: str)

验证

python cli.py storage write .doc/test.json '{"foo":"bar"}'
python cli.py storage read .doc/test.json
# 输出: {"foo":"bar"}

---

第三步：文件锁服务

操作 - 创建 backend/services/file_lock.py

class FileLockService:
    async def acquire_lock(self, file_path: str, agent_id: str) -> bool
    async def release_lock(self, file_path: str, agent_id: str) -> bool
    async def get_locks(self) -> List[LockInfo]
    async def check_locked(self, file_path: str) -> Optional[str]

存储到 .doc/cache/file_locks.json

验证

# CLI 测试
python cli.py lock acquire src/auth/login.py claude-001
# 输出: Lock acquired

python cli.py lock status
# 输出: src/auth/login.py -> claude-001 (2s ago)

python cli.py lock release src/auth/login.py claude-001
# 输出: Lock released

API 测试

curl http://localhost:8000/api/locks
# 输出: {"locks": []}

curl -X POST http://localhost:8000/api/locks/acquire \
  -H "Content-Type: application/json" \
  -d '{"file_path":"src/auth/login.py","agent_id":"claude-001"}'
# 输出: {"acquired":true}

前端对接 - 修改 ResourceMonitorCard.tsx

// 替换 lockedFiles mock
const response = await fetch('http://localhost:8000/api/locks');
const data = await response.json();
setLockedFiles(data.locks);

---

第四步：心跳服务

操作 - 创建 backend/services/heartbeat.py

class HeartbeatService:
    async def update_heartbeat(self, agent_id: str, status: dict) -> None
    async def get_heartbeat(self, agent_id: str) -> Optional[dict]
    async def get_all_heartbeats(self) -> dict
    async def check_timeout(self, timeout_seconds: int) -> List[str]

存储到 .doc/cache/heartbeats.json

验证

# 模拟 Agent 发送心跳
python cli.py heartbeat ping claude-001 '{"status":"working","task":"fixing bug"}'
# 输出: Heartbeat updated

python cli.py heartbeat list
# 输出: claude-001: working (5s ago)

python cli.py heartbeat check-timeout 30
# 输出: No timed out agents

API 测试

curl http://localhost:8000/api/heartbeats
# 输出: {"heartbeats": {"claude-001": {...}}}

---

第五步：Agent 注册服务

操作 - 创建 backend/services/agent_registry.py

class AgentRegistry:
    async def register_agent(self, agent_info: AgentInfo) -> None
    async def get_agent(self, agent_id: str) -> Optional[AgentInfo]
    async def list_agents(self) -> List[AgentInfo]
    async def update_state(self, agent_id: str, state: dict) -> None
    async def get_state(self, agent_id: str) -> Optional[dict]

存储到 .doc/agents/{agent_id}/ 目录

验证

python cli.py agent register claude-001 \
  --name "Claude Code" \
  --role "architect" \
  --model "claude-opus-4.6"
# 输出: Agent registered

python cli.py agent list
# 输出: claude-001 | Claude Code | architect | claude-opus-4.6

python cli.py agent state claude-001 set '{"task":"fixing bug","progress":50}'
python cli.py agent state claude-001 get
# 输出: {"task":"fixing bug","progress":50}

API 测试

curl http://localhost:8000/api/agents
# 输出: {"agents": [...]}

curl http://localhost:8000/api/agents/claude-001/state
# 输出: {"task":"fixing bug","progress":50}

前端对接 - 修改 AgentStatusCard.tsx

// 替换 agents mock
const response = await fetch('http://localhost:8000/api/agents');
const data = await response.json();
setAgents(data.agents);

---

第六步：会议调度器（栅栏同步）

操作 - 创建 backend/services/meeting_scheduler.py

class MeetingScheduler:
    async def wait_for_meeting(self, agent_id: str, meeting_id: str) -> str
    async def get_queue(self, meeting_id: str) -> List[str]
    async def start_meeting(self, meeting_id: str) -> None
    async def end_meeting(self, meeting_id: str) -> None

存储到 .doc/cache/meeting_queue.json

验证

# 终端1 - Agent 1 等待
python cli.py meeting wait design-review claude-001
# 阻塞等待...

# 终端2 - Agent 2 等待
python cli.py meeting wait design-review kimi-002
# 触发会议，输出: Meeting started!

python cli.py meeting queue design-review
# 输出: [claude-001, kimi-002]

API 测试

curl -X POST http://localhost:8000/api/meetings/design-review/wait \
  -H "Content-Type: application/json" \
  -d '{"agent_id":"claude-001"}'
# 输出: {"status":"waiting"}

curl http://localhost:8000/api/meetings/design-review/queue
# 输出: {"waiting_agents":["claude-001"]}

---

第七步：会议记录服务

操作 - 创建 backend/services/meeting_recorder.py

class MeetingRecorder:
    async def create_meeting(self, meeting_info: MeetingInfo) -> str
    async def add_discussion(self, meeting_id: str, agent_id: str, content: str) -> None
    async def update_progress(self, meeting_id: str, step: str) -> None
    async def get_meeting(self, meeting_id: str) -> MeetingInfo
    async def list_meetings(self, date: str) -> List[MeetingInfo]

存储到 .doc/meetings/{YYYY-MM-DD}/{meeting_id}.md

验证

python cli.py meeting create design-review \
  --title "认证方案设计评审" \
  --attendees claude-001,kimi-002,opencode-003
# 输出: Meeting created: design_review_20260304_141000

python cli.py meeting discuss design_review_20260304_141000 \
    --agent claude-001 \
    --content "建议使用 JWT"
# 输出: Discussion added

python cli.py meeting progress design_review_20260304_141000 "收集初步想法"
# 输出: Progress updated

API 测试

curl http://localhost:8000/api/meetings/2026-03-04
# 输出: {"meetings": [...]}

curl http://localhost:8000/api/meetings/design_review_20260304_141000
# 输出: {...}

前端对接 - 修改 MeetingProgressCard.tsx 和 RecentMeetingsCard.tsx

// 替换 steps mock
const response = await fetch('http://localhost:8000/api/meetings/active');
const data = await response.json);
setSteps(data.meeting.steps);

---

第八步：资源管理器（整合锁 + 心跳）

操作 - 创建 backend/services/resource_manager.py

class ResourceManager:
    async def execute_task(self, agent_id: str, task_description: str) -> str
    # 内部逻辑：
    # 1. 解析任务需要的文件
    # 2. 获取所有文件锁
    # 3. 执行任务
    # 4. finally: 释放所有锁

验证

python cli.py execute claude-001 "修改 src/auth/login.py 修复登录 bug"
# 输出: Task executing...
# 输出: Locks acquired: [src/auth/login.py, src/utils/crypto.py]
# 输出: Task completed
# 输出: Locks released

---

第九步：工作流引擎

操作 - 创建 backend/services/workflow_engine.py

class WorkflowEngine:
    async def load_workflow(self, workflow_path: str) -> Workflow
    async def get_next_meeting(self, workflow_id: str) -> Optional[MeetingInfo]
    async def complete_meeting(self, workflow_id: str, meeting_id: str) -> None

验证

python cli.py workflow load .doc/workflow/project_workflow.yaml
# 输出: Workflow loaded: my_project

python cli.py workflow next my_project
# 输出: Next meeting: design_review (attendees: claude-001,kimi-002)

---

第十步：角色分配器（AI 路由）

操作 - 创建 backend/services/role_allocator.py

class RoleAllocator:
    async def allocate_roles(self, task: str, agents: List[str]) -> Dict[str, str]
    # 内部调用 LLM 分析任务，返回角色分配

验证

python cli.py role allocate "重构认证模块" claude-001,kimi-002,opencode-003
# 输出: claude-001 -> architect
# 输出: kimi-002 -> pm
# 输出: opencode-003 -> developer

---

API 接口汇总

路径	方法	说明
/api/agents	GET	获取所有 Agent
/api/agents/{id}/state	GET/POST	Agent 状态
/api/locks	GET	获取所有锁
/api/locks/acquire	POST	获取锁
/api/locks/release	POST	释放锁
/api/heartbeats	GET	获取心跳列表
/api/heartbeats/{id}	POST	更新心跳
/api/meetings/{date}	GET	获取会议列表
/api/meetings/{id}	GET	获取会议详情
/api/meetings/{id}/wait	POST	加入会议等待
/api/workflows/{id}/next	GET	获取下一步

---

前端 Mock 替换顺序

步骤	组件	API 端点
3	ResourceMonitorCard	/api/locks
5	AgentStatusCard	/api/agents
7	MeetingProgressCard	/api/meetings/{id}
7	RecentMeetingsCard	/api/meetings/{date}
-	WorkflowCard	/api/workflows/{id}/next

---

参考开源项目

多智能体协作框架

项目	链接	参考点
AutoGen	github.com/microsoft/autogen	对话驱动协作、异步消息传递、Agent 交接机制
CrewAI	github.com/CrewAIInc/crewAI	角色扮演式 Agent、任务编排、Crew 结构
AgentScope	阿里通义实验室	Agent 编排、安全沙箱、跨框架互操作
MetaGPT	github.com/FoundationAgents/MetaGPT	AI 软件公司、多角色协作、文档生成

FastAPI 项目结构

项目	链接	参考点
FastAPI Project Structure	github.com/brianobot/fastapi_project_structure	分层架构、services/routers/schemas 分离
FastAPI Best Architecture	github.com	伪三层架构、插件系统
FastAPI Tips	github.com	WebSocket 实时通信、性能优化

推荐目录结构：
backend/
├── app/
│   ├── main.py              # FastAPI 入口
│   ├── settings.py          # 配置管理
│   ├── dependencies.py      # 依赖注入
│   ├── middlewares.py       # 中间件
│   ├── api_router.py        # 路由汇总
│   ├── routers/             # API 路由
│   │   ├── agents.py
│   │   ├── locks.py
│   │   └── meetings.py
│   ├── services/            # 业务逻辑
│   │   ├── storage.py
│   │   ├── file_lock.py
│   │   ├── heartbeat.py
│   │   ├── agent_registry.py
│   │   └── meeting_scheduler.py
│   ├── schemas/             # Pydantic 模型
│   │   ├── agent.py
│   │   ├── lock.py
│   │   └── meeting.py
│   └── models/              # 数据模型（如用数据库）
├── cli.py                   # CLI 入口 (Typer)
└── requirements.txt

文件锁实现

库	链接	特点
filelock	github.com/tox-dev/filelock	跨平台、上下文管理器、超时机制
fasteners	github.com/harlowja/fasteners	进程间读写锁

推荐使用 filelock：

from filelock import FileLock, Timeout

lock = FileLock("/path/to/file.lock", timeout=10)
with lock:
    # 临界区代码
    pass

WebSocket 实时更新

FastAPI 原生支持 WebSocket，参考官方文档：fastapi.tiangolo.com/zh/websockets

from fastapi import WebSocket

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    async for data in websocket.iter_text():
        await websocket.send_text(f"Received: {data}")

技术栈推荐

# requirements.txt
fastapi>=0.110.0
uvicorn[standard]>=0.27.0
pydantic>=2.0.0
typer>=0.9.0
aiofiles>=23.0.0
filelock>=3.13.0
pyyaml>=6.0
anthropic>=0.18.0      # Claude API
openai>=1.0.0          # OpenAI API

设计文档参考

文档	链接
AutoGen 编程模型	github.com/microsoft/autogen/docs/design
Agent Worker Protocol	github.com/microsoft/autogen/docs/design