agent_manager.py

# -*- coding: utf-8 -*-
"""
================================================================================
文件名称：agent_manager.py
文件定位：多 Agent 调度中心 / 大模型 API 统一网关
职责说明：
    1. 封装 Kimi2.6 (Moonshot AI) 大模型 API 调用，统一鉴权、超时、重试、日志
    2. 定义 6 个专属 Agent 子类，严格按固定流水线顺序串联执行
    3. 实现 Agent 基类，规范输入输出接口，确保代码与大模型职责完全分离
    4. 总控 AgentManager 负责任务分发、上下文传递、异常熔断、结果聚合
================================================================================
"""

import os
import json
import asyncio
import logging
from typing import Dict, Any, List, Optional, Callable
from datetime import datetime
from dataclasses import dataclass, field

# ------------------------------------------------------------------------------
# 第三方库：OpenAI 兼容接口调用 Kimi2.6（Moonshot API 完全兼容 OpenAI SDK）
# ------------------------------------------------------------------------------
try:
    from openai import AsyncOpenAI
except ImportError:
    AsyncOpenAI = None  # 占位，实际部署时由 requirements.txt 保证安装

# ------------------------------------------------------------------------------
# 日志配置：阿里云部署时可通过日志文件排查问题
# ------------------------------------------------------------------------------
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger("AgentManager")


# ==============================================================================
# 数据模型：Agent 输出标准化结构
# ==============================================================================

@dataclass
class AgentOutput:
    """每个 Agent 处理完毕后的标准化输出结构"""
    agent_name: str
    step: int
    status: str           # "success" | "error"
    result: Dict[str, Any] = field(default_factory=dict)
    raw_response: str = ""    # 大模型原始返回文本（调试用）
    cost_ms: int = 0          # 耗时毫秒
    error_msg: Optional[str] = None


# ==============================================================================
# 大模型客户端封装：统一处理 API Key、BaseURL、重试、超时
# ==============================================================================

class KimiClient:
    """
    Kimi2.6 API 客户端封装（单例模式）
    环境变量依赖：
        - MOONSHOT_API_KEY: 必传，Moonshot AI 官方 API 密钥
        - MOONSHOT_BASE_URL: 可选，默认 https://api.moonshot.cn/v1
    """
    _instance: Optional["KimiClient"] = None
    _lock = asyncio.Lock()

    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self):
        if hasattr(self, "_initialized") and self._initialized:
            return
        self.api_key = os.environ.get("MOONSHOT_API_KEY", "")
        self.base_url = os.environ.get("MOONSHOT_BASE_URL", "https://api.moonshot.cn/v1")
        self.model = "kimi-latest"  # Kimi 2.6 对应模型标识
        self.timeout = 120          # 单次调用超时秒数（学术文本较长）
        self.max_retries = 3        # 失败重试次数
        self.retry_delay = 2.0      # 重试间隔秒数

        if not self.api_key:
            logger.warning("MOONSHOT_API_KEY 未设置！请在阿里云服务器环境变量中配置。")

        if AsyncOpenAI is not None:
            self.client = AsyncOpenAI(
                api_key=self.api_key,
                base_url=self.base_url,
                timeout=self.timeout,
            )
        else:
            self.client = None

        self._initialized = True

    async def chat_completion(
        self,
        messages: List[Dict[str, str]],
        temperature: float = 0.3,
        max_tokens: int = 4096,
    ) -> str:
        """
        调用 Kimi2.6 聊天补全接口，带自动重试与异常熔断
        返回：大模型生成的文本内容
        """
        if self.client is None:
            raise RuntimeError("OpenAI SDK 未安装，请执行 pip install openai")
        if not self.api_key:
            raise RuntimeError("MOONSHOT_API_KEY 未配置，无法调用大模型")

        last_error = None
        for attempt in range(1, self.max_retries + 1):
            try:
                logger.info(f"[Kimi API] 第 {attempt} 次请求，消息轮数: {len(messages)}")
                response = await self.client.chat.completions.create(
                    model=self.model,
                    messages=messages,
                    temperature=temperature,
                    max_tokens=max_tokens,
                )
                content = response.choices[0].message.content.strip()
                logger.info(f"[Kimi API] 请求成功，返回长度: {len(content)} 字符")
                return content

            except Exception as exc:
                last_error = exc
                logger.warning(f"[Kimi API] 第 {attempt} 次请求失败: {exc}")
                if attempt < self.max_retries:
                    await asyncio.sleep(self.retry_delay * attempt)

        raise RuntimeError(f"Kimi API 在 {self.max_retries} 次重试后仍然失败: {last_error}")


# ==============================================================================
# Agent 基类：定义统一接口，6 个子 Agent 全部继承此基类
# ==============================================================================

class BaseAgent:
    """
    Agent 抽象基类
    子类必须实现：
        - name: Agent 名称（英文标识）
        - step: 流水线序号（1~6）
        - system_prompt: 系统级提示词（极窄化约束，防止幻觉）
        - run(user_input: str, context: Dict) -> AgentOutput
    """
    name: str = "BaseAgent"
    step: int = 0
    system_prompt: str = "你是一个智能教育助手。"

    def __init__(self):
        self.kimi = KimiClient()

    async def run(self, user_input: str, context: Dict[str, Any]) -> AgentOutput:
        """
        Agent 执行入口
        参数：
            user_input: 当前步骤需要处理的核心文本（通常是用户原始输入或上游 Agent 输出摘要）
            context: 跨 Agent 传递的上下文字典
        返回：标准化的 AgentOutput
        """
        start_time = datetime.now()
        try:
            # 构造消息列表：系统提示词 + 用户输入
            messages = [
                {"role": "system", "content": self.system_prompt},
                {"role": "user", "content": user_input},
            ]

            # 调用 Kimi2.6 大模型
            raw_response = await self.kimi.chat_completion(messages, temperature=0.3, max_tokens=4096)

            # 解析并结构化结果（子类可覆盖默认解析逻辑）
            structured_result = self._parse_response(raw_response)

            cost_ms = int((datetime.now() - start_time).total_seconds() * 1000)

            return AgentOutput(
                agent_name=self.name,
                step=self.step,
                status="success",
                result=structured_result,
                raw_response=raw_response,
                cost_ms=cost_ms,
            )

        except Exception as exc:
            cost_ms = int((datetime.now() - start_time).total_seconds() * 1000)
            logger.error(f"[{self.name}] 执行失败: {exc}")
            return AgentOutput(
                agent_name=self.name,
                step=self.step,
                status="error",
                result={},
                raw_response="",
                cost_ms=cost_ms,
                error_msg=str(exc),
            )

    def _parse_response(self, raw_response: str) -> Dict[str, Any]:
        """
        默认响应解析器：尝试提取 JSON，失败则返回文本封装
        子类可覆盖此方法以实现更精确的结构化解析
        """
        # 先尝试从 Markdown 代码块中提取 JSON
        cleaned = raw_response.strip()
        if cleaned.startswith("```json"):
            cleaned = cleaned[7:]
        elif cleaned.startswith("```"):
            cleaned = cleaned[3:]
        if cleaned.endswith("```"):
            cleaned = cleaned[:-3]
        cleaned = cleaned.strip()

        try:
            return json.loads(cleaned)
        except json.JSONDecodeError:
            # 非 JSON 格式时，按文本段落封装
            return {
                "analysis_text": raw_response,
                "format": "plain_text",
            }


# ==============================================================================
# 6 个专属 Agent 子类定义（固定流水线顺序 1→2→3→4→5→6）
# ==============================================================================

class Agent1KnowledgeParse(BaseAgent):
    """
    ==============================================================================
    Step 1: 学情解析 Agent
    职责：拆解学习文本，提取核心知识点、标注知识薄弱点、归纳易错点
    输出格式：结构化 JSON（知识点列表、薄弱点列表、易错点列表）
    ==============================================================================
    """
    name = "agent_1_parse"
    step = 1
    system_prompt = """你是一个专业的学情解析专家。你的任务是对学生提交的学习文本进行深度拆解分析。

【核心职责】
1. 提取文本中的核心知识点（至少列出 5 个，每个知识点标注所属学科领域）
2. 识别学生在知识掌握上的薄弱点（至少 3 个，标注薄弱原因：概念模糊/公式记错/逻辑跳跃/审题偏差）
3. 归纳文本中暴露的易错点（至少 3 个，说明错误类型和正确做法）

【输出格式要求】
必须严格按照以下 JSON 格式输出，不要包含任何其他解释文字：
{
    "knowledge_points": [
        {"name": "知识点名称", "domain": "所属学科", "importance": "high/medium/low"}
    ],
    "weak_points": [
        {"name": "薄弱点描述", "reason": "薄弱原因", "suggestion": "改进建议"}
    ],
    "error_prone_points": [
        {"name": "易错点描述", "error_type": "错误类型", "correct_way": "正确做法"}
    ],
    "summary": "整体学情一句话总结"
}

【防幻觉约束】
- 只能基于用户提供的文本内容进行分析，严禁编造文本中不存在的知识点
- 如果文本过短无法判断，请在 summary 中说明"文本信息不足"
- 所有结论必须有文本依据，禁止主观臆测
"""

    def _parse_response(self, raw_response: str) -> Dict[str, Any]:
        # 学情解析 Agent 强制 JSON 输出
        result = super()._parse_response(raw_response)
        # 若解析失败，尝试更鲁棒的兜底策略
        if "analysis_text" in result:
            # 大模型未按 JSON 输出，进行文本提取
            return {
                "knowledge_points": [{"name": "解析异常，原始文本", "domain": "未知", "importance": "medium"}],
                "weak_points": [{"name": "解析异常", "reason": "模型未按 JSON 格式输出", "suggestion": "请检查输入文本"}],
                "error_prone_points": [],
                "summary": result["analysis_text"][:200],
                "parse_warning": "模型输出格式异常，已降级为文本摘要",
            }
        return result


class Agent2LearningProfile(BaseAgent):
    """
    ==============================================================================
    Step 2: 学情画像 Agent
    职责：根据学情解析结果，生成个人学习能力标签、知识短板、学习层级画像
    输出格式：结构化 JSON（能力标签、短板画像、层级评估、成长建议）
    ==============================================================================
    """
    name = "agent_2_profile"
    step = 2
    system_prompt = """你是一个专业的学情画像分析师。你接收学情解析 Agent 的输出结果，为学生生成精准的学习画像。

【核心职责】
1. 生成学习能力标签（至少 4 个标签，如：逻辑推理强、概念理解弱、计算粗心、表达规范等）
2. 绘制知识短板画像（列出知识短板的严重程度分布）
3. 评估当前学习层级（初级/中级/高级，附理由）
4. 给出针对性的成长建议（至少 3 条可执行建议）

【输出格式要求】
必须严格按照以下 JSON 格式输出：
{
    "ability_tags": [
        {"tag": "标签名称", "level": "strong/weak/medium", "evidence": "判断依据"}
    ],
    "shortcoming_profile": {
        "concept_gaps": ["概念缺口1", "概念缺口2"],
        "skill_gaps": ["技能缺口1"],
        "severity": "high/medium/low"
    },
    "learning_level": {
        "level": "初级/中级/高级",
        "score": 75,
        "reason": "评估理由"
    },
    "growth_suggestions": [
        {"action": "具体行动", "target": "预期效果", "duration": "建议周期"}
    ],
    "summary": "画像总结"
}

【防幻觉约束】
- 所有标签必须有学情解析结果作为依据，禁止凭空生成
- 层级评估必须给出量化分数（0-100）和明确理由
- 严禁编造学生个人信息
"""


class Agent3LearningGuide(BaseAgent):
    """
    ==============================================================================
    Step 3: 分层导学 Agent
    职责：定制个性化学习路径、知识点思路引导，只导学、绝不代写作业
    输出格式：结构化 JSON（学习路径、知识点引导、练习建议、资源推荐）
    ==============================================================================
    """
    name = "agent_3_guide"
    step = 3
    system_prompt = """你是一个专业的分层导学导师。你根据学情画像结果，为学生制定个性化学习路径。

【核心职责】
1. 制定分层学习路径（短期/中期/长期目标）
2. 针对薄弱知识点提供思路引导（给出思考方向，不直接给答案）
3. 推荐针对性练习策略（如：错题重做、概念对比、举一反三）
4. 推荐学习资源类型（如：教材章节、视频类型、工具推荐）

【绝对红线】
- 你只提供学习方法和思路引导，严禁直接给出作业答案或代写内容
- 如果用户文本看起来像直接求答案，你必须拒绝并引导其思考
- 所有引导必须以启发式提问或思考框架的形式呈现

【输出格式要求】
必须严格按照以下 JSON 格式输出：
{
    "learning_path": [
        {"phase": "短期(1周内)", "goals": ["目标1", "目标2"], "actions": ["行动1"]}
    ],
    "knowledge_guidance": [
        {"topic": "知识点", "guidance": "思路引导（启发式）", "key_question": "核心思考题"}
    ],
    "practice_strategy": [
        {"method": "练习方法", "description": "具体做法", "frequency": "建议频率"}
    ],
    "resource_recommendation": [
        {"type": "资源类型", "description": "推荐理由"}
    ],
    "warning": "严禁代写作业声明"
}

【防幻觉约束】
- 学习路径必须贴合学生当前层级，不能过难或过易
- 资源推荐必须是通用类型，不能编造不存在的具体书名或链接
- 严禁输出任何可直接复制粘贴的作业答案
"""


class Agent4EthicsCheck(BaseAgent):
    """
    ==============================================================================
    Step 4: 学术伦理风控 Agent
    职责：检测 AI 生成痕迹、学术不端风险、逻辑漏洞、语句不规范问题
    输出格式：结构化 JSON（风险等级、风险项列表、修改建议、合规评分）
    ==============================================================================
    """
    name = "agent_4_ethics"
    step = 4
    system_prompt = """你是一个严格的学术伦理风控专家。你负责检测学生文本中的学术不端风险与质量问题。

【核心职责】
1. 检测 AI 生成痕迹（语言模式同质化、过度 polished、缺乏个人思考痕迹）
2. 识别学术不端风险（抄袭嫌疑、引用不规范、数据捏造迹象）
3. 扫描逻辑漏洞（论证跳跃、因果倒置、以偏概全、循环论证）
4. 检查语句与格式规范（语法错误、标点滥用、段落过长、缺乏过渡）

【输出格式要求】
必须严格按照以下 JSON 格式输出：
{
    "risk_level": "high/medium/low",
    "risk_score": 65,
    "ai_trace": {
        "detected": true/false,
        "indicators": ["指标1", "指标2"],
        "confidence": "high/medium/low"
    },
    "academic_integrity": {
        "issues": [
            {"type": "问题类型", "location": "大致位置", "severity": "high/medium/low", "suggestion": "修改建议"}
        ]
    },
    "logic_flaws": [
        {"type": "漏洞类型", "description": "问题描述", "suggestion": "改进方向"}
    ],
    "language_issues": [
        {"type": "语句问题", "example": "原文片段", "fix": "修改建议"}
    ],
    "overall_suggestion": "总体修改方向"
}

【防幻觉约束】
- AI 痕迹检测只能基于语言特征分析，不能武断判定，必须给出具体指标
- 抄袭嫌疑只能基于文本内部一致性判断，不能查询外部数据库
- 所有风险项必须标注置信度，严禁无中生有制造风险
"""


class Agent5PolishText(BaseAgent):
    """
    ==============================================================================
    Step 5: 学术规范整编 Agent
    职责：全文逻辑润色、段落优化、语句通顺、格式规范化
    输出格式：结构化 JSON（润色后的关键段落、修改对照、格式建议）
    ==============================================================================
    """
    name = "agent_5_polish"
    step = 5
    system_prompt = """你是一个学术规范整编专家。你结合学情解析、画像、导学和风控结果，对学生的原始文本进行规范化整编。

【核心职责】
1. 逻辑润色：优化论证结构，补充过渡句，理顺因果关系
2. 段落优化：拆分过长段落，合并过短段落，统一段落主题
3. 语句通顺：修正语病、消除歧义、统一术语使用
4. 格式规范：统一标题层级、标点使用、引用格式建议

【重要原则】
- 你必须保留学生的原始观点和核心论证，只做规范化整编，不篡改原意
- 润色后的文本必须仍然体现学生个人风格，不能变成成人代笔痕迹
- 如果原始文本存在严重学术不端，你必须在输出中标注而非帮助掩盖

【输出格式要求】
必须严格按照以下 JSON 格式输出：
{
    "polished_fragments": [
        {"original": "原文片段", "polished": "润色后片段", "change_reason": "修改理由"}
    ],
    "structure_suggestions": [
        {"aspect": "结构方面", "current": "现状", "suggested": "建议调整"}
    ],
    "format_guidelines": [
        {"item": "格式项", "standard": "规范标准", "status": "符合/需改进"}
    ],
    "terminology_check": [
        {"term": "术语", "usage_correct": true/false, "suggestion": "使用建议"}
    ],
    "overall_quality_score": 78,
    "summary": "整编总结"
}

【防幻觉约束】
- 润色片段必须基于用户原始文本，不能编造原文中不存在的句子
- 术语检查必须准确，不能误判正确用法
- 质量评分必须给出评分依据
"""


class Agent6GrowthReport(BaseAgent):
    """
    ==============================================================================
    Step 6: 成长档案生成 Agent
    职责：汇总前面所有环节结果，生成完整结构化个人学习成长报告
    输出格式：结构化 JSON + HTML 报告文本（前端可直接渲染）
    ==============================================================================
    """
    name = "agent_6_report"
    step = 6
    system_prompt = """你是一个学习成长档案生成专家。你汇总前 5 个 Agent 的全部分析结果，生成一份完整的个人学习成长报告。

【核心职责】
1. 汇总学情解析结果：核心知识点掌握情况、薄弱点清单
2. 整合学情画像：能力标签云、学习层级、成长建议
3. 收录导学方案：个性化学习路径、知识点引导、练习策略
4. 纳入风控结论：学术伦理风险等级、修改建议
5. 总结整编成果：文本质量评分、结构优化建议
6. 生成综合性成长报告（Markdown/HTML 格式，可直接展示）

【输出格式要求】
必须严格按照以下 JSON 格式输出，report_html 字段包含完整 HTML 报告：
{
    "report_title": "个人学习成长报告",
    "generated_at": "2024-XX-XX",
    "overall_assessment": {
        "learning_level": "初级/中级/高级",
        "risk_level": "high/medium/low",
        "quality_score": 75,
        "key_strengths": ["优势1", "优势2"],
        "key_improvements": ["待改进1", "待改进2"]
    },
    "stage_summary": [
        {"stage": "学情解析", "key_findings": "关键发现摘要"},
        {"stage": "学情画像", "key_findings": "关键发现摘要"},
        {"stage": "分层导学", "key_findings": "关键发现摘要"},
        {"stage": "学术伦理", "key_findings": "关键发现摘要"},
        {"stage": "规范整编", "key_findings": "关键发现摘要"}
    ],
    "growth_plan": {
        "short_term": ["短期行动1", "短期行动2"],
        "medium_term": ["中期目标1"],
        "long_term": ["长期方向1"]
    },
    "report_html": "<html><body>...</body></html>"
}

【HTML 报告要求】
- report_html 字段必须包含完整、格式化的 HTML 文档字符串
- 使用中文标题和清晰的分区（学情概览、能力画像、导学方案、风控结果、成长计划）
- 包含表格展示关键数据，列表展示建议项
- 样式使用内联 CSS，确保前端直接插入 DOM 即可渲染
- 总字数控制在 2000~3000 字以内

【防幻觉约束】
- 报告中的每一项结论都必须能在前 5 个 Agent 的输出中找到依据
- 严禁在汇总时引入新的分析结论或编造数据
- HTML 中的数据必须与 JSON 其他字段完全一致
"""


# ==============================================================================
# AgentManager：总控调度中心，统一管理 6 个 Agent 的注册、排序、执行
# ==============================================================================

class AgentManager:
    """
    总控 Agent 调度器
    职责：
        1. 按固定顺序注册 6 个子 Agent
        2. 接收任务上下文，逐阶段调用 Agent
        3. 管理上下文传递（上游 Agent 输出注入下游 Agent 提示词）
        4. 记录执行日志与耗时，异常时熔断并返回已完成的步骤
    """

    def __init__(self):
        # 固定流水线顺序：1→2→3→4→5→6，顺序绝对不可变更
        self.agents: List[BaseAgent] = [
            Agent1KnowledgeParse(),
            Agent2LearningProfile(),
            Agent3LearningGuide(),
            Agent4EthicsCheck(),
            Agent5PolishText(),
            Agent6GrowthReport(),
        ]
        # 验证顺序
        for i, agent in enumerate(self.agents, start=1):
            if agent.step != i:
                raise RuntimeError(f"Agent 顺序异常: {agent.name} 的 step={agent.step}, 期望 {i}")
        logger.info(f"[AgentManager] 6 个 Agent 已按固定顺序注册完成: {[a.name for a in self.agents]}")

    async def execute_pipeline(
        self,
        input_text: str,
        text_type: str,
        context: Dict[str, Any],
        progress_callback: Optional[Callable[[int, str], None]] = None,
    ) -> Dict[str, Any]:
        """
        执行完整的 6 阶段 Agent 流水线
        参数：
            input_text: 用户原始输入文本
            text_type: 文本类型（homework/note/paper）
            context: 跨步骤上下文字典
            progress_callback: 可选的进度回调函数(step, agent_name)
        返回：
            {
                "step_results": {agent_key: AgentOutput,...},
                "final_context": context,  # 包含全部中间结果
                "success": True/False,
                "completed_steps": int,
                "total_steps": 6,
            }
        """
        step_results: Dict[str, AgentOutput] = {}
        total = len(self.agents)

        for agent in self.agents:
            step = agent.step
            logger.info(f"[AgentManager] 启动 Step {step}: {agent.name}")

            # 进度回调通知（如通知主服务更新任务状态）
            if progress_callback:
                progress_callback(step, agent.name)

            # 构造当前 Agent 的输入：原始文本 + 上游上下文摘要
            user_input = self._build_agent_input(agent, input_text, text_type, context)

            # 执行当前 Agent
            output = await agent.run(user_input, context)
            step_results[agent.name] = output

            # 将结果写入上下文，供下游 Agent 使用
            context[f"{agent.name}_result"] = output.result
            context[f"{agent.name}_raw"] = output.raw_response

            if output.status == "error":
                logger.error(f"[AgentManager] Step {step} 执行失败，流水线中断")
                return {
                    "step_results": step_results,
                    "final_context": context,
                    "success": False,
                    "completed_steps": step - 1,
                    "total_steps": total,
                    "error_step": agent.name,
                    "error_msg": output.error_msg,
                }

            logger.info(f"[AgentManager] Step {step} 完成，耗时 {output.cost_ms}ms")

        # 全部 6 步成功完成
        return {
            "step_results": step_results,
            "final_context": context,
            "success": True,
            "completed_steps": total,
            "total_steps": total,
            "error_step": None,
            "error_msg": None,
        }

    def _build_agent_input(
        self,
        agent: BaseAgent,
        input_text: str,
        text_type: str,
        context: Dict[str, Any],
    ) -> str:
        """
        为每个 Agent 构建专属输入文本
        策略：
            - Agent 1: 原始文本 + 类型说明
            - Agent 2: Agent 1 的 JSON 结果摘要
            - Agent 3: Agent 2 的画像摘要
            - Agent 4: 原始文本 + Agent 3 的导学方案（用于对照）
            - Agent 5: 原始文本 + Agent 4 的风控结果
            - Agent 6: 前 5 个 Agent 的全部结果汇总
        """
        text_type_label = {"homework": "作业", "note": "笔记", "paper": "论文"}.get(text_type, "文本")

        if agent.step == 1:
            return f"【文本类型】{text_type_label}\n\n【学生原文】\n{input_text}\n\n请按 JSON 格式输出学情解析结果。"

        elif agent.step == 2:
            prev = context.get("agent_1_parse_result", {})
            summary = json.dumps(prev, ensure_ascii=False, indent=2)[:3000]
            return f"【学情解析结果】\n{summary}\n\n请基于上述解析结果，生成学生学情画像（JSON 格式）。"

        elif agent.step == 3:
            prev = context.get("agent_2_profile_result", {})
            summary = json.dumps(prev, ensure_ascii=False, indent=2)[:3000]
            return f"【学情画像结果】\n{summary}\n\n请基于画像结果，制定个性化导学方案（JSON 格式）。"

        elif agent.step == 4:
            # 风控 Agent 需要原始文本 + 导学方案（用于检测 AI 痕迹和逻辑问题）
            guide = context.get("agent_3_guide_result", {})
            guide_summary = json.dumps(guide, ensure_ascii=False, indent=2)[:1500]
            return (
                f"【待检测文本】（{text_type_label}）\n{input_text[:4000]}\n\n"
                f"【导学方案参考】\n{guide_summary}\n\n"
                f"请对上述文本进行学术伦理风控检测（JSON 格式）。"
            )

        elif agent.step == 5:
            ethics = context.get("agent_4_ethics_result", {})
            ethics_summary = json.dumps(ethics, ensure_ascii=False, indent=2)[:2000]
            return (
                f"【待整编原文】\n{input_text[:4000]}\n\n"
                f"【风控检测结果】\n{ethics_summary}\n\n"
                f"请对原文进行学术规范整编（JSON 格式）。"
            )

        elif agent.step == 6:
            # 汇总前 5 个 Agent 的全部结果
            summaries = []
            for key in ["agent_1_parse", "agent_2_profile", "agent_3_guide", "agent_4_ethics", "agent_5_polish"]:
                result = context.get(f"{key}_result", {})
                summaries.append(f"【{key} 结果】\n{json.dumps(result, ensure_ascii=False, indent=2)[:1500]}\n")
            combined = "\n".join(summaries)
            return (
                f"【前 5 个 Agent 分析结果汇总】\n{combined}\n\n"
                f"请生成完整的个人学习成长报告（包含 report_html 字段的 JSON 格式）。"
            )

        return input_text