┌─────────────────────────────────────────────────────┐│ Prompt组装引擎（Prompt Assembly Engine）              │├─────────────┬───────────────────────────────────────┤│ 片段注册表   │ 存放所有可复用Prompt片段（md/yaml配置）││ 条件过滤器   │ predicate表达式，动态开关片段          ││ 优先级排序器 │ 控制片段顺序，对抗Lost in the Middle  ││ Token预算器  │ 计数、摘要、滚动窗口、超限保护        ││ 记忆注入器   │ RAG检索、短期对话历史、长期记忆加载   ││ 消息格式化器 │ 转成OpenAI/Anthropic标准message数组   ││ 缓存管理器   │ 静态段标记，支持模型prompt缓存        ││ 组装日志钩子 │ 输出每轮组装报告，用于调试/评估       │└─────────────┴───────────────────────────────────────┘

from dataclasses import dataclassfrom typing import List, Callableimport tiktoken# 1. 定义Prompt片段结构体@dataclassclass PromptSection:    name: str    content: str    priority: int          # 越小越靠前    condition: Callable[[dict], bool]  # 条件判断函数    is_static: bool        # 是否静态可缓存片段class PromptAssemblyEngine:    def __init__(self, token_limit: int = 4096):        self.sections: List[PromptSection] = []        self.token_limit = token_limit        self.enc = tiktoken.get_encoding(\"cl100k_base\")    def register_section(self, sec: PromptSection):        self.sections.append(sec)    def count_tokens(self, text:str) -> int:        return len(self.enc.encode(text))    def assemble(self, runtime_ctx:dict, user_msg:str, history:List[dict]) -> List[dict]:        \"\"\"runtime_ctx：运行时环境，例如 {\"in_git_repo\":True}\"\"\"        # Step1 条件过滤        active_sections = [s for s in self.sections if s.condition(runtime_ctx)]        # Step2 按优先级排序        active_sections.sort(key=lambda x:x.priority)        messages = []        system_parts = []        total_token = 0        # 拼接系统片段        for sec in active_sections:            cnt = self.count_tokens(sec.content)            if total_token + cnt > self.token_limit * 0.85:                break            system_parts.append(sec.content)            total_token += cnt        system_msg = \"\
\
\".join(system_parts)        messages.append({\"role\":\"system\", \"content\":system_msg})        # 追加对话历史（简单截断演示）        hist_token = 0        for turn in reversed(history):            t = self.count_tokens(turn[\"content\"])            if total_token + hist_token + t > self.token_limit:                continue            hist_token += t            messages.insert(1, turn)        # 当前用户消息        messages.append({\"role\":\"user\", \"content\":user_msg})        return messages# ========== 实战使用 ==========if __name__ == \"__main__\":    engine = PromptAssemblyEngine(token_limit=4096)    # 注册片段    engine.register_section(PromptSection(        name=\"agent_identity\",        content=\"你是代码助手Agent，严格遵守输出格式，禁止幻觉。\",        priority=1,        condition=lambda ctx:True,        is_static=True    ))    engine.register_section(PromptSection(        name=\"git_rule\",        content=\"当用户请求代码变更，先查看git状态。\",        priority=10,        condition=lambda ctx: ctx.get(\"in_git_repo\",False),        is_static=True    ))    engine.register_section(PromptSection(        name=\"tool_def\",        content=\"工具列表：read_file, write_file, run_command\",        priority=5,        condition=lambda ctx:True,        is_static=True    ))    runtime_context = {\"in_git_repo\": True}    chat_history = [{\"role\":\"assistant\",\"content\":\"已读取README.md\"}]    prompt_messages = engine.assemble(runtime_context, user_msg=\"帮我修改版本号\", history=chat_history)    for m in prompt_messages:        print(f\"[{m['role']}]: {m['content'][:80]}...\")