Coverage for agentos/prompts/manager.py: 30%
94 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-02 09:59 +0800
1"""
2AgentOS v0.30 Prompt模板管理 — 版本化Prompt仓库。
3支持模板继承、变量注入、A/B测试、回滚。
4"""
6from dataclasses import dataclass, field
7from datetime import datetime
8from typing import Optional
9import json
10import os
13@dataclass
14class PromptTemplate:
15 """Prompt 模板。"""
16 name: str
17 version: str
18 template: str
19 variables: list[str] = field(default_factory=list)
20 description: str = ""
21 parent: str | None = None
22 created_at: str = ""
23 metadata: dict = field(default_factory=dict)
25 def render(self, **kwargs) -> str:
26 """渲染模板，注入变量。"""
27 result = self.template
28 for var in self.variables:
29 value = kwargs.get(var, kwargs.get(f"{{{var}}}", f"{{{{{var}}}}}"))
30 result = result.replace(f"{{{{{var}}}}}", str(value))
31 return result
34class PromptRegistry:
35 """Prompt模板注册中心。"""
37 def __init__(self, storage_path: str = ""):
38 self._templates: dict[str, dict[str, PromptTemplate]] = {}
39 self.storage_path = storage_path
40 if storage_path and os.path.exists(storage_path):
41 self._load()
43 def register(self, template: PromptTemplate):
44 if template.name not in self._templates:
45 self._templates[template.name] = {}
46 template.created_at = template.created_at or datetime.now().isoformat()
47 self._templates[template.name][template.version] = template
48 self._save()
50 def get(self, name: str, version: str = "latest") -> PromptTemplate | None:
51 versions = self._templates.get(name, {})
52 if not versions:
53 return None
54 if version == "latest":
55 sorted_vers = sorted(versions.keys(), key=lambda v: [int(x) for x in v.split(".")])
56 return versions[sorted_vers[-1]]
57 return versions.get(version)
59 def get_version(self, name: str, version: str) -> PromptTemplate | None:
60 return self._templates.get(name, {}).get(version)
62 def list_templates(self) -> list[dict]:
63 result = []
64 for name, versions in self._templates.items():
65 sorted_vers = sorted(versions.keys(), key=lambda v: [int(x) for x in v.split(".")])
66 result.append({
67 "name": name,
68 "versions": sorted_vers,
69 "latest": sorted_vers[-1],
70 "description": versions[sorted_vers[-1]].description,
71 "variables": versions[sorted_vers[-1]].variables,
72 })
73 return result
75 def render(self, name: str, version: str = "latest", **kwargs) -> str:
76 tmpl = self.get(name, version)
77 if not tmpl:
78 raise ValueError(f"Template '{name}' not found")
79 return tmpl.render(**kwargs)
81 def rollback(self, name: str, target_version: str):
82 if name not in self._templates:
83 raise ValueError(f"Template '{name}' not found")
84 versions = self._templates[name]
85 if target_version not in versions:
86 raise ValueError(f"Version '{target_version}' not found for '{name}'")
87 sorted_vers = sorted(versions.keys(), key=lambda v: [int(x) for x in v.split(".")])
88 latest_ver = sorted_vers[-1]
89 if latest_ver == target_version:
90 return
91 # 回滚：基于目标版本创建新版本
92 target = versions[target_version]
93 new_ver_parts = [int(x) for x in latest_ver.split(".")]
94 new_ver_parts[-1] += 1
95 new_ver = ".".join(str(x) for x in new_ver_parts)
96 rolled = PromptTemplate(
97 name=name,
98 version=new_ver,
99 template=target.template,
100 variables=list(target.variables),
101 description=f"Rollback from {latest_ver} to {target_version}",
102 parent=target_version,
103 )
104 self.register(rolled)
105 return rolled
107 def _save(self):
108 if not self.storage_path:
109 return
110 data = {}
111 for name, versions in self._templates.items():
112 data[name] = {}
113 for ver, tmpl in versions.items():
114 data[name][ver] = {
115 "template": tmpl.template,
116 "variables": tmpl.variables,
117 "description": tmpl.description,
118 "parent": tmpl.parent,
119 "created_at": tmpl.created_at,
120 "metadata": tmpl.metadata,
121 }
122 os.makedirs(os.path.dirname(self.storage_path) or ".", exist_ok=True)
123 with open(self.storage_path, "w") as f:
124 json.dump(data, f, indent=2, ensure_ascii=False)
126 def _load(self):
127 with open(self.storage_path) as f:
128 data = json.load(f)
129 for name, versions in data.items():
130 if name not in self._templates:
131 self._templates[name] = {}
132 for ver, vdata in versions.items():
133 self._templates[name][ver] = PromptTemplate(
134 name=name,
135 version=ver,
136 template=vdata["template"],
137 variables=vdata.get("variables", []),
138 description=vdata.get("description", ""),
139 parent=vdata.get("parent"),
140 created_at=vdata.get("created_at", ""),
141 metadata=vdata.get("metadata", {}),
142 )
144 def stats(self) -> dict:
145 total = sum(len(v) for v in self._templates.values())
146 return {"total_templates": len(self._templates), "total_versions": total}
149# ── 预置Prompt模板 ────────────────────────────────
151DEFAULT_PROMPTS = {
152 "agent_system": PromptTemplate(
153 name="agent_system",
154 version="1.0.0",
155 template="""你是一个专业的AI助手，运行在 AgentOS v0.30 上。
156你的任务：{{task}}
157可用工具：{{tools}}
158当前上下文：{{context}}
160请逐步思考并执行。输出格式：
1611. 思考(thinking)
1622. 工具调用(如果需要)
1633. 最终回答""",
164 variables=["task", "tools", "context"],
165 description="Agent核心系统提示",
166 ),
167 "code_review": PromptTemplate(
168 name="code_review",
169 version="1.0.0",
170 template="""审查以下代码。检查维度：{{dimensions}}
172代码：
173```
174{{code}}
175```
177输出结构化报告。""",
178 variables=["code", "dimensions"],
179 description="代码审查Prompt",
180 ),
181 "research_deep": PromptTemplate(
182 name="research_deep",
183 version="1.0.0",
184 template="""对以下主题进行深度调研：{{topic}}
186要求：
1871. 多角度分析（至少3个视角）
1882. 引用权威来源
1893. 对比不同观点
1904. 给出可操作的结论
192调研深度：{{depth}}""",
193 variables=["topic", "depth"],
194 description="深度调研Prompt",
195 ),
196 "summarize": PromptTemplate(
197 name="summarize",
198 version="1.0.0",
199 template="""总结以下内容。摘要长度：{{length}}
201内容：
202{{content}}
204输出格式：{{format}}""",
205 variables=["content", "length", "format"],
206 description="文档摘要Prompt",
207 ),
208 "creative_writing": PromptTemplate(
209 name="creative_writing",
210 version="1.0.0",
211 template="""创作一篇{{genre}}，主题：{{topic}}
212风格：{{style}}
213字数：{{words}}
215要求：
2161. 开头引人入胜
2172. 结构清晰
2183. 结尾有力""",
219 variables=["genre", "topic", "style", "words"],
220 description="创意写作Prompt",
221 ),
222}