🐍 Day10 Practice Validator

"""
Day 10 练习项目验收脚本（行为级验收）
多工具 Agent 综合实战（阶段二收尾）

运行方式：
    python day10_practice_validator.py                  # 验收默认模板文件
    python day10_practice_validator.py --file 你的练习.py

验收方式：
    检查工具注册数量、带重试的工具执行、错误恢复（成功/失败两条路径），
    并用一个"间谍函数"拦截 calculator 调用——验证你的 Agent 是否真的执行了工具。
    空白模板无法通过验收，只有真正完成 TODO 才能得分。
"""

import argparse
import ast
import asyncio
import importlib.util
import inspect
import io
import os
import sys
from contextlib import redirect_stdout
from pathlib import Path

os.system("")  # 让 Windows 终端支持 ANSI 颜色

DEFAULT_FILE = "day10_practice_template.py"
满分 = 100
通过线 = 60
优秀线 = 80


class C:
    GREEN = "\033[92m"
    YELLOW = "\033[93m"
    RED = "\033[91m"
    CYAN = "\033[96m"
    END = "\033[0m"


def 打印(文本, 颜色=None):
    print(f"{颜色}{文本}{C.END}" if 颜色 else 文本)


def 加载学生模块(文件路径: Path):
    spec = importlib.util.spec_from_file_location("学生练习_day10", 文件路径)
    模块 = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(模块)
    return 模块


class 验收器:
    def __init__(self):
        self.总分 = 0

    async def 检查(self, 名称: str, 分值: int, 函数, *参数):
        """运行单个检查；检查函数返回 (得分, 消息)，抛异常记 0 分"""
        打印(f"\n【{名称}】（{分值}分）", C.CYAN)
        try:
            if inspect.iscoroutinefunction(函数):
                得分, 消息 = await asyncio.wait_for(函数(*参数), timeout=60)
            else:
                得分, 消息 = 函数(*参数)
        except asyncio.TimeoutError:
            得分, 消息 = 0, "❌ 检查超时——Agent 可能没有终止条件"
        except Exception as e:
            得分, 消息 = 0, f"❌ {type(e).__name__}: {e}"
        得分 = max(0, min(int(得分), 分值))
        self.总分 += 得分
        颜色 = C.GREEN if 得分 == 分值 else (C.YELLOW if 得分 > 0 else C.RED)
        打印(f"  {消息}  [得分 {得分}/{分值}]", 颜色)


# ==================== 辅助：模拟 Day 9 记忆接口 ====================

class 简易短期记忆:
    def __init__(self):
        self.messages = []

    def add(self, role, content):
        self.messages.append(type("M", (), {"role": role, "content": content})())

    def to_api_format(self):
        return [{"role": m.role, "content": m.content} for m in self.messages]

    def get_recent(self, n=None):
        return self.messages[-n:] if n else self.messages


class 简易RAG:
    def retrieve(self, query, top_k=3):
        return ""


class 简易记忆:
    """提供 Day 9 接口的最小记忆桩：short_memory + rag"""

    def __init__(self):
        self.short_memory = 简易短期记忆()
        self.rag = 简易RAG()


# ==================== 各项检查 ====================

def 检查_工具数量(模块):
    registry = getattr(模块, "TOOLS_REGISTRY", None)
    assert isinstance(registry, dict), "没有找到 TOOLS_REGISTRY"
    assert len(registry) >= 4, (
        f"验收标准要求至少 4 个工具（TODO 2.1/2.2），实际注册了 {len(registry)} 个：{list(registry)}"
    )
    return 20, f"✅ 注册了 {len(registry)} 个工具：{list(registry)}"


async def 检查_工具可调用(模块):
    fn = 模块.TOOLS_REGISTRY.get("calculator", {}).get("function")
    assert fn is not None, "calculator 工具不在注册表中"
    结果 = await asyncio.wait_for(fn(expression="1 + 1"), timeout=15)
    assert "2" in str(结果), f"calculator(1 + 1) 应得 2，实际 {结果!r}"
    return 10, "✅ 工具经注册表包装器（超时+重试）可正常异步调用"


async def 检查_恢复_成功(模块):
    recover = getattr(模块, "execute_tool_with_recovery", None)
    assert callable(recover), "没有找到 execute_tool_with_recovery 函数（TODO 5.1）"
    r = await asyncio.wait_for(
        recover("calculator", {"expression": "3 + 5 * 2"}), timeout=30
    )
    assert r is not None, "成功路径返回 None——TODO 5.1 还没实现"
    assert bool(getattr(r, "success", True)), f"成功路径 success 应为 True，实际 {r!r}"
    assert "13" in str(getattr(r, "output", r)), (
        f"calculator(3 + 5 * 2) 的输出应包含 13，实际 {getattr(r, 'output', r)!r}"
    )
    return 15, f"✅ 成功路径：ToolResult(success=True, output 含 13)"


async def 检查_恢复_失败(模块):
    """不存在的工具：不能崩溃，要把错误作为结果返回给 LLM"""
    recover = getattr(模块, "execute_tool_with_recovery", None)
    r = await asyncio.wait_for(
        recover("不存在的工具xyz", {}), timeout=30
    )
    assert r is not None, "失败路径返回 None——错误应作为结果返回，不能崩溃也不能返回空"
    成功标志 = getattr(r, "success", None)
    if 成功标志 is None and isinstance(r, dict):
        成功标志 = r.get("success")
    if 成功标志 is None and isinstance(r, str):
        # 宽松：返回了包含错误说明的字符串也算"没崩溃"
        assert ("失败" in r or "错误" in r), f"失败路径应返回错误说明，实际 {r!r}"
        return 15, "✅ 失败路径：返回了错误说明（未崩溃）"
    assert 成功标志 is False, (
        f"失败路径 success 应为 False（让 LLM 自行纠正），实际 {r!r}"
    )
    return 15, "✅ 失败路径：不崩溃，返回 success=False 的结果"


async def 检查_Agent回答(模块):
    agent = getattr(模块, "react_agent_with_memory", None)
    assert callable(agent), "没有找到 react_agent_with_memory 函数（TODO 3.1）"
    memory = 简易记忆()
    tools = list(模块.TOOLS_REGISTRY.values())
    缓冲 = io.StringIO()
    with redirect_stdout(缓冲):
        回答 = await asyncio.wait_for(
            agent("帮我计算 3 + 5 * 2", tools, memory, max_steps=6, stream=False),
            timeout=50,
        )
    assert isinstance(回答, str) and 回答.strip(), f"应返回字符串回答，实际 {回答!r}"
    assert 回答.strip() != "最终回答", (
        "返回的还是模板占位符'最终回答'——TODO 3.1 主循环还没实现"
    )
    return 10, f"✅ Agent 返回了真实回答：{回答[:50]!r}"


async def 检查_Agent真调工具(模块):
    """用间谍函数拦截 calculator：验证 Agent 的回答真的来自工具执行"""
    agent = getattr(模块, "react_agent_with_memory", None)
    原包装 = 模块.TOOLS_REGISTRY["calculator"]["function"]
    调用记录 = {"n": 0}

    async def 间谍(*args, **kwargs):
        调用记录["n"] += 1
        return await 原包装(*args, **kwargs)

    模块.TOOLS_REGISTRY["calculator"]["function"] = 间谍
    try:
        memory = 简易记忆()
        tools = list(模块.TOOLS_REGISTRY.values())
        缓冲 = io.StringIO()
        with redirect_stdout(缓冲):
            await asyncio.wait_for(
                agent("帮我计算 3 + 5 * 2", tools, memory, max_steps=6, stream=False),
                timeout=50,
            )
    finally:
        模块.TOOLS_REGISTRY["calculator"]["function"] = 原包装

    assert 调用记录["n"] >= 1, (
        f"Agent 没有通过工具注册表执行 calculator（拦截到 {调用记录['n']} 次调用）。"
        "工具执行必须走 TOOLS_REGISTRY 里包装好的函数，而不是绕过它"
    )
    return 20, f"✅ 间谍函数拦截到 calculator 被真实执行了 {调用记录['n']} 次"


def 检查_类型注解(文件路径):
    树 = ast.parse(文件路径.read_text(encoding="utf-8"))
    函数们 = [n for n in ast.walk(树) if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef))]
    有注解 = [f for f in 函数们 if f.returns is not None]
    assert len(有注解) >= 4, f"至少 4 个函数应有返回值类型注解，目前 {len(有注解)}/{len(函数们)}"
    return 5, f"✅ {len(有注解)}/{len(函数们)} 个函数有返回值注解"


# ==================== 主流程 ====================

async def 运行验收(验收员, 模块, 文件路径):
    await 验收员.检查("工具系统：至少 4 个工具", 20, 检查_工具数量, 模块)
    await 验收员.检查("工具包装器：可异步调用", 10, 检查_工具可调用, 模块)
    await 验收员.检查("错误恢复：成功路径", 15, 检查_恢复_成功, 模块)
    await 验收员.检查("错误恢复：失败路径不崩溃", 15, 检查_恢复_失败, 模块)
    await 验收员.检查("Agent：返回真实回答", 10, 检查_Agent回答, 模块)
    await 验收员.检查("Agent：真的执行了工具（间谍拦截）", 20, 检查_Agent真调工具, 模块)
    await 验收员.检查("类型注解", 5, 检查_类型注解, 文件路径)


def main():
    parser = argparse.ArgumentParser(description="Day 10 练习验收脚本（行为级验收）")
    parser.add_argument("--file", default=DEFAULT_FILE, help="练习文件路径（默认 %(default)s）")
    args = parser.parse_args()
    文件路径 = Path(args.file)

    打印("=" * 60)
    打印("📋 Day 10 练习验收 —— 多工具 Agent 综合实战")
    打印("=" * 60)
    打印(f"\n📂 验收文件：{文件路径}")

    if not 文件路径.exists():
        打印(f"\n❌ 文件不存在：{文件路径}", C.RED)
        print("请先完成 day10_practice_template.py 中的 TODO，或用 --file 指定你的练习文件")
        sys.exit(1)

    打印("\n【加载练习代码】（5分）", C.CYAN)
    加载分 = 5
    try:
        模块 = 加载学生模块(文件路径)
        打印("  ✅ 代码加载成功，没有语法错误", C.GREEN)
    except Exception as e:
        加载分 = 0
        模块 = None
        打印(f"  ❌ 加载失败：{type(e).__name__}: {e}", C.RED)
        print("  💡 请先修复语法/运行错误再验收")

    验收员 = 验收器()
    验收员.总分 += 加载分
    if 模块 is not None:
        asyncio.run(运行验收(验收员, 模块, 文件路径))

    总分 = 验收员.总分
    打印("\n" + "=" * 60)
    if 总分 >= 优秀线:
        打印(f"  总分：{总分}/{满分}   🎉 验收通过！阶段二正式毕业！", C.GREEN)
    elif 总分 >= 通过线:
        打印(f"  总分：{总分}/{满分}   ⚠️ 基本通过，建议按上面的提示完善", C.YELLOW)
    else:
        打印(f"  总分：{总分}/{满分}   ❌ 未通过——先完成模板中标红的 TODO 再来验收", C.RED)
    打印("=" * 60)


if __name__ == "__main__":
    main()