# 只用标准库
"""三个 Agent 设计模式的最小实现（离线可运行）：

1. Plan-and-Execute + 失败重新规划
2. Skills 渐进式加载：启动时只把技能的名字和描述放进上下文，需要时再加载全文
3. 上下文预算：超过预算时把早期对话压缩成摘要，保留系统提示词和最近几轮

模型部分用规则模拟，接口形状和真实做法一致；真实项目中 plan / summarize 由大模型完成。
运行：python patterns.py
"""
from __future__ import annotations

import re
from dataclasses import dataclass, field
from pathlib import Path

# ---------------- 1. Plan-and-Execute ----------------
TOOLS = {
    "query_sales": lambda a: {"rows": [("华北", 120), ("华东", 200)]} if a.get("month") else {"error": "缺少 month 参数"},
    "make_chart": lambda a: {"chart": f"bar({len(a['rows'])} bars)"},
    "send_report": lambda a: {"sent_to": a["to"]},
}


@dataclass
class Step:
    tool: str
    args: dict
    status: str = "pending"       # pending / done / failed
    result: dict | None = None


def plan(goal: str, feedback: str | None = None) -> list[Step]:
    """规划器（真实项目中由大模型输出结构化计划）。收到失败反馈时修正计划。"""
    month = re.search(r"(\d{4}-\d{2})", goal)
    args = {"month": month.group(1)} if (month and feedback) else {}
    return [Step("query_sales", args), Step("make_chart", {}), Step("send_report", {"to": "ops@example.com"})]


def plan_and_execute(goal: str, max_replans: int = 2) -> tuple[list[Step], list[str]]:
    log, feedback = [], None
    for attempt in range(max_replans + 1):
        steps = plan(goal, feedback)
        log.append(f"第 {attempt + 1} 次规划：{[s.tool for s in steps]}")
        ctx: dict = {}
        for s in steps:
            args = {**s.args, **({"rows": ctx["rows"]} if s.tool == "make_chart" else {})}
            s.result = TOOLS[s.tool](args)
            if "error" in s.result:
                s.status = "failed"
                feedback = f"{s.tool} 失败：{s.result['error']}"
                log.append(feedback)
                break
            s.status = "done"
            ctx.update(s.result)
        else:
            return steps, log
    return steps, log + ["超过最大重新规划次数，转人工"]


# ---------------- 2. Skills 渐进式加载 ----------------
@dataclass
class Skill:
    name: str
    description: str
    path: Path

    def body(self) -> str:
        text = self.path.read_text(encoding="utf-8")
        return text.split("---", 2)[2].strip()          # 去掉 frontmatter


def discover_skills(root: Path) -> list[Skill]:
    skills = []
    for f in sorted(root.glob("*/SKILL.md")):
        meta = dict(re.findall(r"^(name|description):\s*(.+)$", f.read_text(encoding="utf-8"), re.M))
        skills.append(Skill(meta["name"], meta["description"], f))
    return skills


def skills_index(skills: list[Skill]) -> str:
    """放进系统提示词的只有这一小段索引。"""
    return "\n".join(f"- {s.name}: {s.description}" for s in skills)


def pick_skill(question: str, skills: list[Skill]) -> Skill | None:
    """真实项目中由模型根据索引决定加载哪个技能（或通过一个 load_skill 工具）。"""
    keywords = {"refund-policy": ["退款", "退货"], "sql-report": ["统计", "报表"]}
    return next((s for s in skills if any(k in question for k in keywords.get(s.name, []))), None)


# ---------------- 3. 上下文预算 ----------------
def n_tokens(text: str) -> int:
    return len(text)                    # 粗略估算：中文约 1 字 1 token；真实项目用模型的分词器


@dataclass
class Context:
    system: str
    budget: int
    keep_recent: int = 4
    messages: list[tuple[str, str]] = field(default_factory=list)
    summary: str = ""

    def add(self, role: str, text: str) -> None:
        self.messages.append((role, text))
        while self.size() > self.budget and len(self.messages) > self.keep_recent:
            old = self.messages[: len(self.messages) - self.keep_recent]
            self.messages = self.messages[len(self.messages) - self.keep_recent:]
            self.summary = summarize(self.summary, old)

    def size(self) -> int:
        return n_tokens(self.system) + n_tokens(self.summary) + sum(n_tokens(t) for _, t in self.messages)

    def render(self) -> list[tuple[str, str]]:
        head = [("system", self.system)] + ([("system", "之前对话摘要：" + self.summary)] if self.summary else [])
        return head + self.messages


def summarize(prev: str, msgs: list[tuple[str, str]]) -> str:
    """真实项目中由模型生成摘要，要求保留：用户目标、已确认的事实、未完成事项。"""
    facts = [t for r, t in msgs if r == "user" and re.search(r"\d|地址|改", t)]
    if not facts:
        return prev
    return (prev + "；" if prev else "") + "用户提供的信息：" + "；".join(facts)[:60]


if __name__ == "__main__":
    steps, log = plan_and_execute("生成 2026-09 各大区销售报表并发给运营")
    print("\n".join(log))
    print("最终步骤状态：", [(s.tool, s.status) for s in steps])
    assert [s.status for s in steps] == ["done", "done", "done"] and len(log) == 3

    skills = discover_skills(Path(__file__).parent / "skills")
    print("系统提示词中的技能索引：\n" + skills_index(skills))
    s = pick_skill("我上周买的耳机想退款，800 元", skills)
    print("按需加载技能：", s.name, "->", s.body().splitlines()[0])
    assert s.name == "refund-policy" and "500" in s.body()

    ctx = Context(system="你是售后助手。", budget=40)
    for i, t in enumerate(["我的订单号是 A1001", "好的", "收货地址改到上海", "嗯",
                           "什么时候发货", "预计明天", "能开发票吗", "可以"]):
        ctx.add("user" if i % 2 == 0 else "assistant", t)
    print("压缩后上下文：", ctx.render())
    assert ctx.size() <= 40 or len(ctx.messages) == ctx.keep_recent
    assert "A1001" in ctx.summary