LeetCode Agent Lab - 多智能体系统设计
多智能体系统（src/multi_agent/）是项目的核心业务模块。它将 LeetCode 解题拆解为 5 个专业化角色，通过编排器（Orchestrator）协调工作。本文深入分析其架构设计、协议层、评分系统和上下文压缩等关键技术。
1. 整体架构
多智能体系统采用 编排器 + 专业化 Agent 的架构：
┌──────────────────────┐
│ MultiAgentOrchestrator │
│ (orchestrator.py) │
└──────────┬─────────────┘
│ 调度
┌─────────┬──────────┼──────────┬──────────┐
▼ ▼ ▼ ▼ ▼
SolverAgent Reviewer TestAgent ScoreAgent NoteWriter
(生成代码) (审查代码) (扩展测试) (评分) (总结沉淀)编排器的核心逻辑在 orchestrator.py（约 960 行），负责：
- 加载题目，查询 RAG 记忆
- 并行运行多个 Solver 配置（baseline / optimized / conservative）
- 对每个候选方案：生成 → 执行基础测试 → 审查 → 修改 → 测试循环 → 评分
- 投票选出最优方案
- NoteWriter 生成模板卡片，持久化到 SQLite
2. 五个 Agent 角色
2.1 SolverAgent — 代码生成
SolverAgent 根据不同配置文件（profile）生成候选代码：
class SolverAgent(BaseAgent):
def solve(
self,
problem: Problem,
solver_profile: str = "baseline",
retrieved_memories: list[RetrievedMemory] | None = None,
) -> SolutionDraft:
response = self.backend.chat_json(
system_prompt=load_prompt("solver_system"),
user_prompt=build_solver_user_prompt(problem, solver_profile, retrieved_memories),
temperature=self.temperature,
)
return parse_solver_response(response, solver_profile=solver_profile)设计要点：
- 支持多种 solver profile（baseline、optimized、conservative），通过配置区分提示词策略
- 返回值经过
parse_solver_response统一解析为SolutionDraftPydantic 模型 - 可注入历史记忆（
retrieved_memories），让 Solver 借鉴过去的解题经验
2.2 ReviewerAgent — 代码审查
ReviewerAgent 审查 Solver 生成的代码，判断是否需要修改，并生成测试扩展请求：
- 输入：题目 + 候选代码 + 执行结果
- 输出：ReviewResult（decision: pass/revise + issues 列表 + TestGenerationRequest）
2.3 TestAgent — 扩展测试
TestAgent 根据 Reviewer 的请求生成额外测试用例，执行后返回测试结果。它不只是生成测试——还会在发现失败后触发 reassess（重新评估） 流程。
2.4 ScoreAgent — LLM 评分
ScoreAgent 提供基于 LLM 的代码评分，与规则评分融合后得到最终分数。
2.5 NoteWriterAgent — 总结沉淀
NoteWriterAgent 为最终胜出的方案生成模板卡片（template card）和反思（reflection），存入 SQLite 记忆库，供后续解题参考。
3. 协议层设计
3.1 为什么要协议层
多 Agent 系统中，每个 Agent 的输出格式不同。如果让编排器直接解析每个 Agent 的原始 JSON，代码会变得极其脆弱——任何 Agent 输出格式变化都会波及整个系统。
协议层的作用是：统一约束输出格式，集中管理解析逻辑。
3.2 schema_version 策略
所有 Agent 输出强制携带 schema_version: "v1"：
{
"schema_version": "v1",
"code": "def two_sum(nums, target): ...",
"approach": "哈希表",
"time_complexity": "O(n)"
}版本策略是 严格写入、宽松读取：
- 写入时：必须是已知版本，否则拒绝
- 读取时：未知版本降级处理，不直接报错
这样既能保证数据质量，又能在升级时保持向后兼容。
3.3 集中解析
所有协议解析集中在 protocols/ 子包中，通过 Pydantic model_validate 统一反序列化：
# protocols/parser.py 伪代码
def parse_solver_response(raw: dict, solver_profile: str) -> SolutionDraft:
raw.setdefault("schema_version", "v1")
if raw["schema_version"] != "v1":
raise ProtocolParseError(f"Unsupported schema: {raw['schema_version']}")
return SolutionDraft.model_validate(raw)4. 评分系统
4.1 规则评分
规则评分由 scoring/scorer.py 中的纯函数实现，基于四个维度加权：
| 维度 | 权重 | 评估方式 |
|---|---|---|
| 正确性 (correctness) | 35% | 测试通过率 |
| 复杂度 (complexity) | 25% | complexity_to_score() 映射 |
| 健壮性 (robustness) | 25% | 运行时错误、失败用例数 |
| 可读性 (readability) | 15% | 行数、行长、命名风格 |
复杂度映射函数：
def complexity_to_score(expr: str) -> float:
text = (expr or "").strip().lower().replace(" ", "")
if "o(1)" in text:
return 100.0
if "o(logn)" in text or "o(log" in text:
return 95.0
if "o(n)" in text and "nlogn" not in text and "n^" not in text:
return 88.0
if "o(nlogn)" in text:
return 78.0
if re.search(r"o\(n\^?2\)", text):
return 60.0
if re.search(r"o\(n\^?3\)", text):
return 40.0
return 65.0可读性评分：
def readability_score(code: str) -> float:
lines = [line for line in code.splitlines() if line.strip()]
score = 90.0
line_count = len(lines)
max_len = max(len(line) for line in lines)
if line_count > 80:
score -= 24
elif line_count > 50:
score -= 14
elif line_count > 35:
score -= 8
if max_len > 140:
score -= 18
elif max_len > 110:
score -= 10
# 驼峰命名混用扣分
snake_case_hits = len(re.findall(r"\b[a-z]+(_[[a-z0-9]+)+\b", code))
camel_hits = len(re.findall(r"\b[a-z]+[A-Z][A-Za-z0-9]*\b", code))
if camel_hits > snake_case_hits * 2:
score -= 6
return clamp(score, 0.0, 100.0)4.2 评分融合
规则评分和 Agent 评分通过可配置的 alpha 系数融合：
最终分 = alpha × agent分 + (1 - alpha) × 规则分默认 alpha = 0.6，置信度阈值 0.65。当 Agent 评分置信度不足时，回退到纯规则评分。
4.3 投票选优
每个候选方案获得一个 VoteResult，最终按加权投票选出胜者：
| 维度 | 权重 |
|---|---|
| 正确性 | 40% |
| 效率 | 20% |
| 审查风险 | 20% |
| 简洁性 | 10% |
| 稳定性 | 10% |
5. AOP 上下文压缩
5.1 问题：上下文爆炸
在 reassess（重新评估）阶段，需要把扩展测试的执行结果传回 Reviewer。但如果测试用例很多，完整的执行结果会占用大量 token，导致：
- API 调用成本飙升
- 上下文窗口被挤占
- LLM 注意力被稀释
5.2 解决方案：只传统计 + Top-K 失败案例
AOP 层在 reassess 前对上下文进行压缩：
原始上下文（可能数千 token）
↓ 压缩
统计摘要 + top-k 失败用例摘要（数百 token）具体做法：
- 保留：总用例数、通过数、失败数、通过率
- 保留：失败用例中 top-k 的输入/期望/实际输出摘要
- 丢弃：通过用例的详细输出、完整执行日志
这样既保证 Reviewer 有足够的信息做判断，又不会浪费 token。
6. 存储与记忆
6.1 SQLite 存储
io/storage.py 使用 SQLite 持久化四类数据：
| 表 | 作用 |
|---|---|
runs | 每次运行的完整记录 |
notes | NoteWriter 生成的模板卡片和反思 |
scorecards | 评分卡 |
memories | 记忆条目（供 RAG 检索） |
6.2 记忆检索
记忆检索使用 token 重叠评分（词袋模型），当没有重叠时回退到时间最近优先：
# 简化逻辑
def retrieve_memories(query: str, top_k: int) -> list[RetrievedMemory]:
query_tokens = set(tokenize(query))
candidates = storage.get_all_memories()
scored = []
for mem in candidates:
mem_tokens = set(tokenize(mem.template_card + mem.reflection))
overlap = len(query_tokens & mem_tokens)
scored.append((overlap, mem))
scored.sort(key=lambda x: (-x[0], x[1].created_at))
return [mem for _, mem in scored[:top_k]]7. 编排流程完整时序
1. 加载题目 → 查询记忆
2. 并行启动 N 个 Solver（不同 profile）
├── Solver(profile="baseline") → code_v1
├── Solver(profile="optimized") → code_v2
└── Solver(profile="conservative") → code_v3
3. 对每个候选：
├── 执行基础测试 → run_result
├── Reviewer 审查 → review_result
├── 如果 review.decision == "revise":
│ ├── 应用修改 → revised_code
│ └── 测试循环（最多 K 轮）：
│ ├── TestAgent 生成扩展测试
│ ├── 执行扩展测试
│ └── Reviewer reassess（AOP 压缩上下文）
├── 评分：规则分 + Agent 分 → fused_score
└── 生成 VoteResult
4. 投票选优 → winner
5. NoteWriter 生成模板卡片 → 持久化到 SQLite
6. 返回 ProblemRunSummary总结
| 知识点 | 要点 |
|---|---|
| 架构 | 编排器 + 5 个专业化 Agent |
| 协议层 | schema_version: "v1"，严格写入、宽松读取 |
| 评分 | 规则评分（4 维度）+ Agent 评分，alpha 融合 |
| 投票 | 正确性 40% + 效率 20% + 审查风险 20% + 简洁 10% + 稳定 10% |
| AOP 压缩 | reassess 阶段只传统计 + top-k 失败案例 |
| 存储 | SQLite 四表：runs / notes / scorecards / memories |
| 记忆检索 | token 重叠评分 + 时间回退 |
项目源码：leetcode-coder-agent