Context Engineering

from dataclasses import dataclass
import json, time
@dataclass
class KnowledgeChunk:
    source: str          # 来源标识（可点开的链接或文件路径）
    updated_at: str      # 更新时间：模型据此判断时效
    tier: str            # fact（可溯源事实）| inference（推断）| opinion（观点）
    text: str
    score: float = 0.0   # 相关性得分，用于排序与截断
def render_knowledge(chunks: list[KnowledgeChunk], budget_chars: int = 6000) -> str:
    """
    按相关性排序、按预算截断，并且**每一条都带来源与层级**。
    来源不是装饰，它是模型判断「该不该相信」的依据。
    """
    if not chunks:
        return "（本次未检索到相关资料。若信息不足，请明确说明而不是猜测。）"
    ordered = sorted(chunks, key=lambda c: -c.score)
    lines, used = [], 0
    for c in ordered:
        block = (f"[{c.tier}] {c.text.strip()}\n"
                 f"    来源：{c.source}　更新时间：{c.updated_at}")
        if used + len(block) > budget_chars:
            break                      # 硬性截断，防单条超长资料挤爆预算
        lines.append(block)
        used += len(block)
    dropped = len(ordered) - len(lines)
    tail = f"\n（另有 {dropped} 条相关资料因预算限制未纳入。）" if dropped else ""
    return "\n\n".join(lines) + tail
def build_messages(
    instructions: str,
    tool_defs: list[dict],
    task: str,
    progress_summary: str,
    knowledge: list[KnowledgeChunk],
    new_input: str,
) -> list[dict]:
    """
    严格按「稳定度递减」排列，保证前缀缓存命中率最大化。
    注意：instructions 与 tool_defs 里绝对不能出现时间戳等动态内容。
    """
    stable_head = json.dumps(tool_defs, ensure_ascii=False, sort_keys=True)
    return [
        {"role": "system", "content": instructions},                     # ① 最稳定
        {"role": "system", "content": f"可用工具：\n{stable_head}"},      # ② 顺序固定
        {"role": "user", "content": f"任务目标：{task}"},                 # ④ 一轮内不变
        {"role": "user", "content": f"已完成步骤：\n{progress_summary}"},  # ⑤ 压缩后
        {"role": "user", "content": f"参考资料：\n{render_knowledge(knowledge)}"},  # ③/⑤
        {"role": "user", "content": new_input},                          # ⑥ 变化最频繁，放最后
    ]
def summarize_progress(steps: list[dict]) -> str:
    """
    阶段摘要必须结构化。自由文本摘要会越摘越糊。
    固定四段：目标 / 已完成 / 关键结论 / 待办
    """
    done = [s for s in steps if s.get("status") == "ok"]
    failed = [s for s in steps if s.get("status") not in ("ok", None)]
    return (
        f"· 已完成（{len(done)} 步）："
        + "；".join(f"{s['tool']}({s.get('brief','')})" for s in done[-8:])
        + f"\n· 失败/放弃（{len(failed)} 步）："
        + "；".join(f"{s['tool']}→{s.get('status')}" for s in failed[-5:])
        + "\n· 待办：见任务目标未覆盖的部分"
    )