Merge pull request #478 from SHDB108/feat/prompt-auto-designer

feat(prompt): add auto prompt designer helper

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
+```

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. 若信息缺失，可做合理假设，但需保证提示词安全、可执行。

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 的数值，表示优化后对原目标的匹配度，越高越好。

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"]

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)

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)