多 Agent 协作刷题：一个出题、一个解答、一个评测的分工模式

一、单一 Agent 刷题的局限：什么都做，什么都做不好

目前大多数 AI 刷题工具都是单 Agent 模式：一个模型负责从出题到解答到评测的所有环节。这个架构在简单题上没有问题，但当题目变难时，单一 Agent 的上下文窗口和推理能力都会成为瓶颈。

具体来说，单 Agent 刷题有三个核心问题。第一，角色混淆。同一个 Agent 既要"像一个考官一样出题"，又要"像一个学生一样解题"，还要"像一个评测系统一样打分"——这三种角色的思维模式是冲突的。第二，自证偏差。Agent 出的题，Agent 自己解答，Agent 自己评测——这个过程缺乏外部验证。第三，能力上限。出题需要创造力，解答需要推理力，评测需要严谨的判断力——单一模型的这三项能力不可能同时是最优的。

多 Agent 协作是解决这些问题的自然思路。将刷题系统分解为三个独立的 Agent：出题 Agent、解题 Agent 和评测 Agent，各司其职，互相制约。

flowchart TB
    A[用户: 设定训练目标] --> B[出题 Agent]
    B --> |"根据用户薄弱点生成题目"| C[题目生成]
    C --> D[解题 Agent]
    D --> |"分析题目并生成解答"| E[解题过程]
    E --> F[评测 Agent]
    F --> |"验证正确性 + 评估代码质量"| G{通过?}
    
    G -->|是| H[记录到用户知识库]
    G -->|否| I[反馈给解题 Agent]
    I --> D
    
    H --> J[更新用户能力画像]
    J --> A
    
    subgraph 通讯协议
        K["出题Agent → 解题Agent: {题目, 约束, 预期难度}"]
        L["解题Agent → 评测Agent: {代码, 推理链, 复杂度分析}"]
        M["评测Agent → 出题Agent: {正确性, 质量分, 建议题目方向}"]
    end

二、三个 Agent 的角色定义与协作协议

出题 Agent（Problem Generator）： 负责根据用户的能力画像和薄弱点，生成针对性练习题目。它的输入是用户的刷题记录（哪些题型做得好、哪些容易卡住、最近刷的频率），输出是一个结构化的题目描述（问题陈述、输入输出格式、约束条件、预期难度）。

出题 Agent 的核心能力需求：理解算法题的结构和难度分层、能创建有效的变体题目（同一模式不同表述）、以及根据用户反馈动态调整题目难度。

解题 Agent（Solution Solver）： 负责解答出题 Agent 生成的题目。它需要完整的算法能力——读懂题目、选择算法、编码实现、分析复杂度。解题 Agent 的关键特征是：它不知道"标准答案"（因为题目是动态生成的），这避免了自证偏差。

评测 Agent（Judge Evaluator）： 负责验证解题 Agent 的代码是否正确，并给出质量评分。它不是简单地检查输出是否匹配预期——它要做的是：运行测试用例、分析代码的边界条件处理、评估时间空间复杂度、以及给出优化建议。

评测 Agent 的一个关键特性是：它同时收到题目和解答，但它的判断逻辑不应受任何 Agent 影响。这类似于区块链中的"共识节点"——它的权威来自于算法的客观性，而非任何一个 Agent 的背书。

三、多 Agent 协作的工程框架实现

"""
多 Agent 刷题协作引擎
核心设计：
1. 三个 Agent 通过消息队列异步通信
2. 评测 Agent 有沙箱执行能力，确保代码安全
3. 结果通过评分卡反馈给出题 Agent，形成自适应出题
"""
from dataclasses import dataclass, field
from typing import List, Optional, Dict
from enum import Enum
import json


class AgentRole(Enum):
    """Agent 角色"""
    PROBLEM_GENERATOR = "出题Agent"
    SOLUTION_SOLVER = "解题Agent"
    JUDGE_EVALUATOR = "评测Agent"


class Difficulty(Enum):
    EASY = "简单"
    MEDIUM = "中等"
    HARD = "困难"


@dataclass
class Problem:
    """一道算法题"""
    problem_id: str
    title: str
    description: str
    input_format: str
    output_format: str
    constraints: str
    difficulty: Difficulty
    tags: List[str]
    test_cases: List[dict]  # [{"input": "...", "output": "..."}]
    expected_complexity: str  # 预期的最优复杂度


@dataclass
class Solution:
    """一份题解"""
    problem_id: str
    code: str
    language: str
    reasoning_chain: List[str]  # 推理步骤
    time_complexity: str
    space_complexity: str


@dataclass
class Evaluation:
    """一次评测结果"""
    solution: Solution
    passed_cases: int
    total_cases: int
    time_used_ms: float
    memory_used_mb: float
    verdict: str  # AC/WA/TLE/MLE/RE
    code_quality_score: float  # 0-10
    suggestions: List[str]
    needs_rework: bool


class MultiAgentOrchestrator:
    """
    多 Agent 编排器
    负责三个 Agent 之间的任务调度和消息传递。
    
    为什么用编排器模式：三个 Agent 不应直接互相调用，
    而是通过编排器解耦。这样任意一个 Agent 的模型可以
    独立替换，不影响其他 Agent。
    """
    
    def __init__(self, 
                 problem_agent,   # 出题 Agent 的 LLM 客户端
                 solver_agent,    # 解题 Agent 的 LLM 客户端
                 judge_agent):    # 评测 Agent 的 LLM 客户端
        self.problem_agent = problem_agent
        self.solver_agent = solver_agent
        self.judge_agent = judge_agent
        
        # 记录每个用户的刷题数据
        self.user_profiles: Dict[str, dict] = {}
        # 记录已生成过的题目（去重用）
        self.generated_problems: set = set()
    
    def run_session(self, user_id: str, 
                    focus_areas: List[str],
                    session_size: int = 3) -> List[dict]:
        """
        运行一次完整的刷题会话
        
        focus_areas: 用户想重点训练的方向
        session_size: 本次会话的题目数量
        返回：每题的结果摘要
        
        为什么需要 focus_areas：出题 Agent 需要知道用户薄弱点，
        生成有针对性的题目。
        """
        user_profile = self.user_profiles.get(user_id, {})
        results = []
        
        for i in range(session_size):
            # 步骤 1：出题 Agent 生成题目
            problem = self._generate_problem(
                user_profile=user_profile,
                focus_areas=focus_areas,
                session_index=i
            )
            
            if problem is None:
                continue
            
            # 步骤 2：解题 Agent 生成解答
            solution = self._solve_problem(problem)
            
            # 步骤 3：评测 Agent 验证
            evaluation = self._evaluate_solution(problem, solution)
            
            # 步骤 4：更新用户画像
            self._update_profile(user_id, problem, evaluation)
            
            results.append({
                "problem": problem,
                "solution": solution,
                "evaluation": evaluation,
                "metrics": {
                    "passed": evaluation.passed_cases,
                    "total": evaluation.total_cases,
                    "quality": evaluation.code_quality_score
                }
            })
            
            # 步骤 5：反馈循环——如果需要，重新解答
            if evaluation.needs_rework and i < session_size:
                # 给解题 Agent 反馈后重试一次
                feedback = "\n".join(evaluation.suggestions)
                solution = self._solve_problem(problem, feedback)
                evaluation = self._evaluate_solution(problem, solution)
                results[-1] = {
                    "problem": problem,
                    "solution": solution,
                    "evaluation": evaluation
                }
        
        return results
    
    def _generate_problem(self, user_profile: dict,
                          focus_areas: List[str],
                          session_index: int) -> Optional[Problem]:
        """调用出题 Agent 生成题目"""
        prompt = f"""
你是一个算法出题专家。请根据以下信息生成一道算法题：

用户能力画像：{json.dumps(user_profile, ensure_ascii=False)}
训练方向：{focus_areas}
当前是本次会话的第 {session_index + 1} 题

要求：
1. 题目难度递增，第 1 题偏简单
2. 题目描述清晰，约束条件自洽
3. 提供 3 组测试用例（含边界条件）
4. 输出 JSON 格式的题目数据

为什么要求约束条件自洽：如果条件矛盾（如 n<=10 但要求 O(1)），
评测 Agent 将无法正确判断。
"""
        try:
            response = self.problem_agent.complete(prompt)
            problem_data = json.loads(response)
            
            # 去重检查
            problem_id = problem_data.get("problem_id", "")
            if problem_id in self.generated_problems:
                return None  # 重复题目跳过
            self.generated_problems.add(problem_id)
            
            return Problem(**problem_data)
        except Exception as e:
            return None
    
    def _solve_problem(self, problem: Problem,
                       feedback: str = "") -> Solution:
        """调用解题 Agent 生成解答"""
        prompt = f"""
请解答以下算法题：

{problem.description}
输入格式：{problem.input_format}
输出格式：{problem.output_format}
约束条件：{problem.constraints}

{"上次解答的反馈：" + feedback if feedback else ""}

请提供：
1. 解题思路（逐步推理）
2. 完整代码（带注释）
3. 时间复杂度分析
4. 空间复杂度分析
"""
        response = self.solver_agent.complete(prompt)
        
        # 从响应中提取代码（简化处理）
        return Solution(
            problem_id=problem.problem_id,
            code=self._extract_code(response),
            language="python",
            reasoning_chain=self._extract_reasoning(response),
            time_complexity="O(n)",
            space_complexity="O(1)"
        )
    
    def _evaluate_solution(self, problem: Problem, 
                           solution: Solution) -> Evaluation:
        """调用评测 Agent 验证解答"""
        # 首先在沙箱中运行测试用例
        passed, total, verdict, time_ms, memory_mb = \
            self._run_test_cases(solution.code, problem.test_cases)
        
        # 然后让评测 Agent 给代码质量评分
        quality_score = self._assess_code_quality(
            problem.description, solution.code, verdict)
        
        suggestions = self._generate_suggestions(
            verdict, quality_score, solution)
        
        return Evaluation(
            solution=solution,
            passed_cases=passed,
            total_cases=total,
            time_used_ms=time_ms,
            memory_used_mb=memory_mb,
            verdict=verdict,
            code_quality_score=quality_score,
            suggestions=suggestions,
            needs_rework=(passed < total or quality_score < 6)
        )
    
    def _run_test_cases(self, code: str, 
                        test_cases: List[dict]) -> tuple:
        """在安全沙箱中执行代码"""
        import subprocess
        passed = 0
        max_time = 0
        max_memory = 0
        
        for tc in test_cases:
            try:
                process = subprocess.run(
                    ["python3", "-c", self._wrap_for_test(code, tc)],
                    capture_output=True, timeout=5, text=True
                )
                if process.stdout.strip() == tc["output"].strip():
                    passed += 1
            except:
                pass
        
        total = len(test_cases)
        verdict = "AC" if passed == total else "WA"
        return passed, total, verdict, max_time, max_memory
    
    def _assess_code_quality(self, description: str, 
                              code: str, verdict: str) -> float:
        """评测代码质量"""
        prompt = f"""
请对以下代码的代码质量打分（0-10分）：

题目：{description}
评测结果：{verdict}
代码：
{code}

评分维度：可读性、命名规范、算法选择、边界处理
"""
        response = self.judge_agent.complete(prompt)
        try:
            return float(response.strip().split()[0])
        except:
            return 5.0
    
    def _update_profile(self, user_id: str, problem: Problem, 
                        evaluation: Evaluation):
        """更新用户能力画像"""
        profile = self.user_profiles.get(user_id, {
            "total_solved": 0,
            "ac_rate": 0.0,
            "weak_areas": [],
            "strength_areas": [],
            "recent_difficulty": "Easy"
        })
        
        profile["total_solved"] += 1
        if evaluation.verdict == "AC":
            profile["ac_rate"] = (
                (profile["ac_rate"] * (profile["total_solved"] - 1) + 1) 
                / profile["total_solved"]
            )
        
        self.user_profiles[user_id] = profile
    
    def _extract_code(self, response: str) -> str:
        # 从 LLM 响应中提取代码块
        pass
    
    def _extract_reasoning(self, response: str) -> List[str]:
        pass
    
    def _wrap_for_test(self, code: str, test_case: dict) -> str:
        pass
    
    def _generate_suggestions(self, verdict: str, quality: float,
                              solution: Solution) -> List[str]:
        pass

四、Agent 协作的边界与可靠性分析

Agent 间的通信开销。 三个 Agent 之间每次交互都需要一次 LLM 调用，加上可能的代码执行调用。一次完整的"出题 → 解答 → 评测"周期可能需要 3-5 次 LLM 调用。如果再加上"反馈 → 重新解答"的循环，调用次数可能更多。对于实时刷题场景，需要评估用户等待时间的可接受性。

Agent 能力的互补性。 如果三个 Agent 使用同一个基础模型，它们的能力瓶颈是相同的——这并没有解决单 Agent 的"能力上限"问题。真正有价值的做法是：为不同角色使用专门优化的模型。出题 Agent 可以用代码生成能力强的模型（如 CodeLlama），解题 Agent 可以用推理能力强的模型（如 DeepSeek-R1），评测 Agent 可以用逻辑严谨的模型。

错误传播与自愈。 如果出题 Agent 生成了一道有 bug 的题目（如约束条件矛盾），评测 Agent 在测试时应该能发现这个问题。但如果评测 Agent 也没有发现，整个系统的输出就是错误的。解决方案是引入交叉验证机制——解题 Agent 也可以对题目的合理性做反馈。

成本与收益的平衡。 三个 Agent 意味着 3 倍的 LLM 调用成本。对于个人使用的刷题工具，这个成本可能偏高。但对于教育平台或比赛场景，高质量的多轮验证是值得的。

五、总结

多 Agent 协作将刷题系统从"一个全能者"变成了"一个团队"。出题 Agent 保证题目质量，解题 Agent 保证多种思路，评测 Agent 保证结果的准确性。三个 Agent 相互制约、相互补充，形成了一个闭环质量保障体系。

这个架构的核心思想不是"三个比一个好"，而是"分工让每个 Agent 能专注于自己的职责"。当每个 Agent 只需要做好一件事时，整个系统的质量上限就由每个 Agent 的深度（而非广度）决定——而单 Agent 模式的质量上限受限于广度与深度的矛盾。

未来，当 Agent 能力进一步增强时，我们可能看到更多角色的加入——如"解析 Agent"（专门用于理解用户提问）、"总结 Agent"（将解题过程提炼为方法论）——刷题工具将逐步从"自动化题解"走向"智能化教练"。

# 算法
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

深入理解 AI Agent · MEMORY #02：记忆的工程机制

avatar AI Agent技术社区
cover

DeepSeek V4 Pro正式版上线：AI Agent能力跃升背后的行业趋势

avatar AI Agent技术社区
cover

炸穿 GitHub[特殊字符]！DeepSeek 开源神器封神：全员插件[特殊字符]、狂省 Token[特殊字符]，AI Agent 居然能自己改自己[特殊字符]

avatar AI Agent技术社区