diff --git a/PROMPT_AUTO_DESIGNER_CHANGES.md b/PROMPT_AUTO_DESIGNER_CHANGES.md new file mode 100644 index 00000000..872e62e7 --- /dev/null +++ b/PROMPT_AUTO_DESIGNER_CHANGES.md @@ -0,0 +1,12 @@ +# Prompt Auto Designer 改动说明 + +- 新增 `PromptAutoDesigner` 服务,封装 prompt 生成/优化的核心流程,并通过定制的异常、数据模型和 LLM 调用链路提高易用性。 +- 引入 `generator_cn.yaml`、`optimizer_cn.yaml` 两个模板,分别用于从场景输入生成新 prompt 以及对既有 prompt 做迭代优化。 +- 扩展 `agentuniverse/prompt/__init__.py` 以便直接从 `agentuniverse.prompt` 导入新服务,同时补充了针对成功/失败路径的单元测试。 +- 当前文档部分暂未调整,保留上游版本;如需接入,可在后续 PR 中扩展使用指南。 + +## 测试命令 + +```bash +PYTHONPATH=$PWD poetry run pytest tests/test_agentuniverse/unit/prompt/test_prompt_auto_designer.py -q +``` diff --git a/agentuniverse/base/util/prompt/auto_prompt/generator_cn.yaml b/agentuniverse/base/util/prompt/auto_prompt/generator_cn.yaml new file mode 100644 index 00000000..1df6fa94 --- /dev/null +++ b/agentuniverse/base/util/prompt/auto_prompt/generator_cn.yaml @@ -0,0 +1,52 @@ +metadata: + version: auto_prompt.generator_cn +introduction: | + 你是一位资深的智能体 Prompt 架构专家,擅长把业务需求转化为高质量的提示词。 +target: | + 根据输入信息,为一个三段式(introduction、target、instruction)的智能体 Prompt 设计内容。 +instruction: | + 请阅读以下上下文,并输出可直接使用的 Prompt 设计: + + 场景描述: + {scenario} + + 任务目标: + {objective} + + 目标受众 / 角色设定: + {audience} + + 语气与风格要求: + {tone} + + 生成语言: + {language} + + 智能体可获取的输入信号: + {inputs} + + 期望交付的输出形式: + {outputs} + + 必须遵循的约束: + {constraints} + + 背景/知识补充: + {context} + + 可参考的示例或历史片段: + {examples} + + 输出要求: + 1. 结果必须是合法 JSON,严格遵循以下结构且不要包含额外文字: + {{ + "introduction": "...", + "target": "...", + "instruction": "...", + "rationale": "...", + "suggested_variables": ["...", "..."] + }} + 2. introduction、target、instruction 三段均需使用 {language} 撰写,并与场景强相关。 + 3. rationale 简洁说明设计思路与关键考量。 + 4. suggested_variables 明确列出在 instruction 中需要替换的变量名称(例如 "background"、"chat_history")。 + 5. 若信息缺失,可做合理假设,但需保证提示词安全、可执行。 diff --git a/agentuniverse/base/util/prompt/auto_prompt/optimizer_cn.yaml b/agentuniverse/base/util/prompt/auto_prompt/optimizer_cn.yaml new file mode 100644 index 00000000..0dc0e9d6 --- /dev/null +++ b/agentuniverse/base/util/prompt/auto_prompt/optimizer_cn.yaml @@ -0,0 +1,45 @@ +metadata: + version: auto_prompt.optimizer_cn +introduction: | + 你是负责 Prompt 质量评审与优化的专家,擅长找出改进点并提供高质量修订方案。 +target: | + 基于现有 Prompt,给出改进后的三段式版本,并说明优化理由。 +instruction: | + 当前 Prompt 内容如下(保持原格式): + {current_prompt} + + 场景补充: + {scenario} + + 预期目标: + {objective} + + 已知问题或痛点: + {issues} + + 成功衡量指标: + {success_metrics} + + 风格/语气要求: + {style} + + 生成语言: + {language} + + 额外背景与边界: + {context} + + 输出要求: + 1. 仅输出合法 JSON,不要添加解释性文字,结构如下: + {{ + "introduction": "...", + "target": "...", + "instruction": "...", + "rationale": "...", + "change_log": ["...", "..."], + "score": 0-100 + }} + 2. introduction、target、instruction 采用 {language},保持语义清晰并覆盖现有问题。 + 3. rationale 简要说明整体优化策略与风险提示。 + 4. change_log 用条目指出关键修改点,每项都以动词开头。 + 5. score 为 0-100 的数值,表示优化后对原目标的匹配度,越高越好。 diff --git a/agentuniverse/prompt/__init__.py b/agentuniverse/prompt/__init__.py index 4048bef5..d5009b62 100644 --- a/agentuniverse/prompt/__init__.py +++ b/agentuniverse/prompt/__init__.py @@ -4,3 +4,7 @@ # @Author : heji # @Email : lc299034@antgroup.com # @FileName: __init__.py + +from agentuniverse.prompt.prompt_auto_designer import PromptAutoDesigner + +__all__ = ["PromptAutoDesigner"] diff --git a/agentuniverse/prompt/prompt_auto_designer.py b/agentuniverse/prompt/prompt_auto_designer.py new file mode 100644 index 00000000..5b2e57cc --- /dev/null +++ b/agentuniverse/prompt/prompt_auto_designer.py @@ -0,0 +1,230 @@ +# !/usr/bin/env python3 +# -*- coding:utf-8 -*- +"""小工具,用来把结构化输入快速拼成可用的 prompt。""" +from __future__ import annotations + +import json +import re +from typing import Any, Callable, Dict, List, Optional + +from pydantic import BaseModel, Field + +from agentuniverse.base.component.component_enum import ComponentEnum +from agentuniverse.base.util.prompt_util import generate_template +from agentuniverse.llm.llm import LLM +from agentuniverse.llm.llm_manager import LLMManager +from agentuniverse.prompt.prompt import Prompt +from agentuniverse.prompt.prompt_manager import PromptManager +from agentuniverse.prompt.prompt_model import AgentPromptModel + + +class PromptAutoDesignerError(RuntimeError): + """LLM 掉链子或者结果不对时就丢这个异常。""" + + +class PromptGenerationRequest(BaseModel): + """生成 prompt 时最常用的那几块素材就靠它承载。""" + + scenario: str = Field(..., description="核心业务或应用场景。") + objective: str = Field(..., description="该 prompt 需要驱动智能体完成的目标。") + audience: Optional[str] = Field(default=None, description="面向的用户或风格要求。") + tone: Optional[str] = Field(default=None, description="希望的语气与表述风格。") + language: str = Field(default="中文", description="生成 prompt 所使用的语言。") + inputs: List[str] = Field(default_factory=list, description="智能体将接收的关键输入。") + outputs: List[str] = Field(default_factory=list, description="期望的输出内容形态。") + constraints: List[str] = Field(default_factory=list, description="必须遵循的约束或限制条件。") + additional_context: Optional[str] = Field(default=None, description="可选的背景信息、知识库说明等。") + examples: Optional[str] = Field(default=None, description="示例问答或既有 prompt 片段。") + + +class PromptOptimizationRequest(BaseModel): + """草稿已经有了,需要再雕琢就填这个模型。""" + + prompt: AgentPromptModel = Field(..., description="当前使用的 prompt 三段文案。") + scenario: Optional[str] = Field(default=None, description="补充的场景描述。") + objective: Optional[str] = Field(default=None, description="该 prompt 需要实现的目标。") + issues: List[str] = Field(default_factory=list, description="当前 prompt 遇到的问题或痛点。") + success_metrics: List[str] = Field(default_factory=list, description="衡量优化效果的指标。") + style: Optional[str] = Field(default=None, description="目标语气或写作风格。") + language: str = Field(default="中文", description="优化后 prompt 的语言。") + additional_context: Optional[str] = Field(default=None, description="补充的上下文或边界。") + + +class PromptDesignResult(BaseModel): + """生成阶段的产出,带上 prompt 文本和变量建议。""" + + prompt: AgentPromptModel + prompt_text: str + rationale: Optional[str] = None + suggested_variables: List[str] = Field(default_factory=list) + + +class PromptOptimizationResult(BaseModel): + """优化后的结果,附带变更记录和评分。""" + + prompt: AgentPromptModel + prompt_text: str + rationale: Optional[str] = None + change_log: List[str] = Field(default_factory=list) + score: Optional[float] = None + + +class PromptAutoDesigner: + """主流程:拉模板、调 LLM、解析结果,全都打包在这。""" + + def __init__( + self, + generation_prompt_version: str = "auto_prompt.generator_cn", + optimization_prompt_version: str = "auto_prompt.optimizer_cn", + llm_name: Optional[str] = None, + llm_factory: Optional[Callable[[], LLM]] = None, + ): + self.generation_prompt_version = generation_prompt_version + self.optimization_prompt_version = optimization_prompt_version + self.llm_name = llm_name + self._llm_factory = llm_factory + + def generate_prompt(self, request: PromptGenerationRequest) -> PromptDesignResult: + """把输入塞进模板,让 LLM 现写一份合适的 prompt。""" + payload = self._build_generation_payload(request) + raw = self._invoke_llm(self.generation_prompt_version, payload) + parsed = self._parse_json(raw) + prompt_model = self._build_prompt_model(parsed) + prompt_text = generate_template(prompt_model, ["introduction", "target", "instruction"]).strip() + return PromptDesignResult( + prompt=prompt_model, + prompt_text=prompt_text, + rationale=self._safe_get(parsed, "rationale"), + suggested_variables=self._ensure_list(parsed.get("suggested_variables")), + ) + + def optimize_prompt(self, request: PromptOptimizationRequest) -> PromptOptimizationResult: + """基于旧 prompt 打补丁,返回改进版本。""" + payload = self._build_optimization_payload(request) + raw = self._invoke_llm(self.optimization_prompt_version, payload) + parsed = self._parse_json(raw) + prompt_model = self._build_prompt_model(parsed, fallback=request.prompt) + prompt_text = generate_template(prompt_model, ["introduction", "target", "instruction"]).strip() + score = parsed.get("score") + normalized_score = self._coerce_float(score) if score is not None else None + return PromptOptimizationResult( + prompt=prompt_model, + prompt_text=prompt_text, + rationale=self._safe_get(parsed, "rationale"), + change_log=self._ensure_list(parsed.get("change_log")), + score=normalized_score, + ) + + def _invoke_llm(self, prompt_version: str, payload: dict[str, Any]) -> str: + prompt = self._get_prompt(prompt_version) + llm = self._resolve_llm() + chain = prompt.as_langchain() | llm.as_langchain_runnable() + try: + return chain.invoke(payload) + except Exception as exc: + raise PromptAutoDesignerError( + f"LLM 调用失败,prompt_version={prompt_version}, payload_keys={list(payload.keys())}" + ) from exc + + def _resolve_llm(self) -> LLM: + if self._llm_factory: + llm = self._llm_factory() + else: + name = self.llm_name or "__default_instance__" + llm = LLMManager().get_instance_obj(name) + if llm is None: + raise PromptAutoDesignerError("没有找到可用的 LLM 配置,请先在应用配置中注册默认模型。") + if llm.component_type != ComponentEnum.LLM: + raise PromptAutoDesignerError("llm_factory 返回的实例类型不正确。") + return llm + + def _get_prompt(self, version: str) -> Prompt: + prompt = PromptManager().get_instance_obj(version) + if prompt is None: + raise PromptAutoDesignerError(f"未注册 prompt 版本:{version}") + return prompt.create_copy() + + def _build_generation_payload(self, request: PromptGenerationRequest) -> dict[str, str]: + return { + "scenario": request.scenario, + "objective": request.objective, + "audience": request.audience or "未指定", + "tone": request.tone or "专业、清晰", + "language": request.language, + "inputs": self._format_bullets(request.inputs), + "outputs": self._format_bullets(request.outputs), + "constraints": self._format_bullets(request.constraints, fallback="暂无额外约束"), + "context": request.additional_context or "无额外背景信息", + "examples": request.examples or "暂无示例", + } + + def _build_optimization_payload(self, request: PromptOptimizationRequest) -> dict[str, str]: + current_prompt = generate_template(request.prompt, ["introduction", "target", "instruction"]).strip() + return { + "current_prompt": current_prompt or "当前提示词为空", + "scenario": request.scenario or "未提供", + "objective": request.objective or "未提供", + "issues": self._format_bullets(request.issues, fallback="暂无明确问题"), + "success_metrics": self._format_bullets(request.success_metrics, fallback="暂无成功指标"), + "style": request.style or "保持内容准确、结构化的语气", + "language": request.language, + "context": request.additional_context or "无补充背景", + } + + @staticmethod + def _build_prompt_model(parsed: Dict[str, Any], fallback: Optional[AgentPromptModel] = None) -> AgentPromptModel: + data = { + "introduction": parsed.get("introduction"), + "target": parsed.get("target"), + "instruction": parsed.get("instruction"), + } + seed = AgentPromptModel(**data) + if fallback: + seed = seed + fallback + if not seed: + raise PromptAutoDesignerError("LLM 响应未提供有效的 prompt 内容。") + return seed + + @staticmethod + def _ensure_list(value: Any) -> List[str]: + if value is None: + return [] + if isinstance(value, list): + return [str(item) for item in value] + return [str(value)] + + @staticmethod + def _format_bullets(values: List[str], fallback: str = "无") -> str: + if not values: + return fallback + return "\n".join(f"- {value}" for value in values) + + @staticmethod + def _coerce_float(value: Any) -> Optional[float]: + if isinstance(value, (int, float)): + return float(value) + if isinstance(value, str): + cleaned = re.findall(r"[-+]?[0-9]*\.?[0-9]+", value) + if cleaned: + return float(cleaned[0]) + return None + + @staticmethod + def _parse_json(raw: str) -> Dict[str, Any]: + try: + return json.loads(raw) + except json.JSONDecodeError: + match = re.search(r"\{.*}", raw, flags=re.DOTALL) + if match: + try: + return json.loads(match.group()) + except json.JSONDecodeError: + pass + raise PromptAutoDesignerError("LLM 响应不是有效的 JSON,请检查提示词模板或模型输出。") + + @staticmethod + def _safe_get(payload: Dict[str, Any], key: str) -> Optional[str]: + value = payload.get(key) + if value is None: + return None + return str(value) diff --git a/tests/test_agentuniverse/unit/prompt/test_prompt_auto_designer.py b/tests/test_agentuniverse/unit/prompt/test_prompt_auto_designer.py new file mode 100644 index 00000000..57ddc765 --- /dev/null +++ b/tests/test_agentuniverse/unit/prompt/test_prompt_auto_designer.py @@ -0,0 +1,108 @@ +import json + +import pytest + +from agentuniverse.prompt.prompt_auto_designer import ( + PromptAutoDesigner, + PromptAutoDesignerError, + PromptGenerationRequest, + PromptOptimizationRequest, +) +from agentuniverse.prompt.prompt_model import AgentPromptModel + + +def test_generate_prompt_success(monkeypatch): + captured: dict = {} + + def fake_invoke(self, version, payload): + captured["version"] = version + captured["payload"] = payload + return json.dumps( + { + "introduction": "你是企业知识库的智能体助手。", + "target": "帮助客服在三步内给出准确答案。", + "instruction": "始终读取 background 并结合 input 给出结论。", + "rationale": "针对客服流程强调输入来源。", + "suggested_variables": ["background", "input"], + } + ) + + monkeypatch.setattr(PromptAutoDesigner, "_invoke_llm", fake_invoke) + + designer = PromptAutoDesigner() + request = PromptGenerationRequest( + scenario="企业在线客服机器人", + objective="快速回答常见问题并引用 FAQ 数据", + audience="客服专员", + tone="友好且专业", + language="中文", + inputs=["用户提问", "FAQ 检索结果"], + outputs=["结构化回答", "引用来源"], + constraints=["回答前先确认信息来源", "拒绝超出知识库范围的请求"], + additional_context="机器人部署在官网,需要适配文字与语音双渠道。", + examples="Q: 如何重置密码?\nA: 请前往账号设置...", + ) + + result = designer.generate_prompt(request) + + assert result.prompt.introduction == "你是企业知识库的智能体助手。" + assert result.prompt.target.startswith("帮助客服") + assert result.prompt_text.startswith("你是企业知识库的智能体助手。") + assert result.suggested_variables == ["background", "input"] + assert result.rationale == "针对客服流程强调输入来源。" + assert captured["version"] == "auto_prompt.generator_cn" + assert "- 用户提问" in captured["payload"]["inputs"] + + +def test_optimize_prompt_merges_fallback(monkeypatch): + base_prompt = AgentPromptModel( + introduction="你是一名财务分析助手。", + target="帮助分析季度营收表现。", + instruction="阅读背景信息并回答财务问题。", + ) + + def fake_invoke(self, version, payload): + assert version == "auto_prompt.optimizer_cn" + assert "季度营收" in payload["current_prompt"] + return json.dumps( + { + "introduction": "你是一名上市公司财报分析顾问。", + "instruction": "优先列出关键财务指标,再给出风险提示。", + "rationale": "强化指标顺序并加入风险提醒。", + "change_log": ["优化身份描述", "补充风险提示步骤"], + "score": "92.5", + } + ) + + monkeypatch.setattr(PromptAutoDesigner, "_invoke_llm", fake_invoke) + + designer = PromptAutoDesigner() + request = PromptOptimizationRequest( + prompt=base_prompt, + scenario="上市公司财报解读", + objective="总结营收并指出风险", + issues=["未明确风险提示", "回答顺序不稳定"], + success_metrics=["输出需覆盖收入、利润、风险三部分"], + style="专业且有条理", + ) + + result = designer.optimize_prompt(request) + + assert result.prompt.introduction == "你是一名上市公司财报分析顾问。" + assert result.prompt.target == "帮助分析季度营收表现。" + assert result.prompt.instruction.startswith("优先列出关键财务指标") + assert result.change_log == ["优化身份描述", "补充风险提示步骤"] + assert result.score == pytest.approx(92.5, rel=1e-3) + assert result.rationale == "强化指标顺序并加入风险提醒。" + + +def test_generate_prompt_invalid_json(monkeypatch): + monkeypatch.setattr(PromptAutoDesigner, "_invoke_llm", lambda self, version, payload: "not-json") + designer = PromptAutoDesigner() + request = PromptGenerationRequest( + scenario="安防巡检机器人", + objective="生成巡检指令", + ) + + with pytest.raises(PromptAutoDesignerError): + designer.generate_prompt(request)