147 lines
5.9 KiB
Python
147 lines
5.9 KiB
Python
# 只用标准库
|
||
"""三个 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
|