结构化提示词的工程化实践

从自由文本到工程化模板：构建可维护、可测试、可复用的提示词体系

---

为什么需要结构化提示词

自由文本提示词的三大痛点：

1. 不可预测：同一个意图的不同表述，LLM 可能给出完全不同质量的输出
2. 不可维护：提示词散落在代码各处，修改一个提示词需要搜索整个代码库
3. 不可复用：每个场景都从头写提示词，没有积累和沉淀

结构化提示词的目标：将提示词从"自然语言技巧"转变为"软件工程实践"。

自由文本提示词：
  "帮我写一个函数，要求..."（随意、不可控、不可复用）

结构化提示词：
  <context>项目上下文</context>
  <task>具体任务</task>
  <constraints>约束条件</constraints>
  <output_format>输出格式</output_format>
  （可预测、可维护、可复用）

---

一、XML 标签结构化

为什么用 XML 标签

XML 标签是当前最被推荐的提示词结构化方式，原因：

• 边界清晰：标签明确界定了每个部分的开始和结束
• 层级表达：支持嵌套，可以表达复杂的结构
• 模型友好：主流 LLM（Claude、GPT-4）都能很好地解析 XML 结构
• 工具兼容：可以用 XML 解析器程序化处理

基础模板

<system>
你是一个{role}。

<context>
{background_information}
</context>

<task>
{task_description}
</task>

<constraints>
- {constraint_1}
- {constraint_2}
- {constraint_3}
</constraints>

<output_format>
{format_specification}
</output_format>

<examples>
<example>
<input>{example_input}</input>
<output>{example_output}</output>
</example>
</examples>
</system>

实际应用示例：代码审查

<system>
你是一个高级代码审查专家。

<context>
项目：Python Web 应用（FastAPI + PostgreSQL）
代码规范：PEP 8 + 项目自定义规则
审查重点：安全性、性能、可维护性
</context>

<task>
审查以下代码变更，输出结构化的审查意见。
</task>

<code_diff>
{diff_content}
</code_diff>

<review_criteria>
<criterion name="security" weight="high">
SQL 注入、XSS、CSRF、认证/授权绕过、敏感信息泄露
</criterion>
<criterion name="performance" weight="medium">
N+1 查询、未使用索引、内存泄漏、不必要的计算
</criterion>
<criterion name="maintainability" weight="medium">
命名规范、函数长度、职责单一、测试覆盖
</criterion>
<criterion name="correctness" weight="high">
逻辑错误、边界情况、竞态条件、错误处理
</criterion>
</review_criteria>

<output_format>
对每个发现，输出：
- severity: critical / major / minor / suggestion
- category: security / performance / maintainability / correctness
- file: 文件路径
- line: 行号
- description: 问题描述
- suggestion: 修复建议
- code_snippet: 建议的代码修改
</output_format>
</system>

---

二、JSON Schema 约束输出

结构化输出的核心思路

通过定义精确的 JSON Schema，强制 LLM 输出特定结构的数据。

# 方法 1：Prompt 中声明 Schema
SCHEMA_PROMPT = """
请按照以下 JSON Schema 输出结果：

{
  "type": "object",
  "properties": {
    "summary": {
      "type": "string",
      "description": "100字以内的摘要"
    },
    "key_findings": {
      "type": "array",
      "items": {
        "type": "object",
        "properties": {
          "finding": {"type": "string"},
          "confidence": {"type": "number", "minimum": 0, "maximum": 1},
          "evidence": {"type": "string"}
        },
        "required": ["finding", "confidence"]
      }
    },
    "recommendation": {
      "type": "string",
      "enum": ["proceed", "revise", "reject"]
    }
  },
  "required": ["summary", "key_findings", "recommendation"]
}

仅输出 JSON，不要包含其他文字。
"""

# 方法 2：使用 API 的 structured output 功能
from pydantic import BaseModel, Field

class ReviewFinding(BaseModel):
    finding: str = Field(description="发现的问题")
    confidence: float = Field(ge=0, le=1, description="置信度")
    evidence: str = Field(description="支撑证据")

class CodeReviewResult(BaseModel):
    summary: str = Field(max_length=200, description="审查摘要")
    key_findings: list[ReviewFinding] = Field(description="关键发现")
    recommendation: str = Field(
        description="建议",
        json_schema_extra={"enum": ["proceed", "revise", "reject"]}
    )

# OpenAI Structured Output
response = client.beta.chat.completions.parse(
    model="gpt-4o",
    messages=[{"role": "user", "content": review_prompt}],
    response_format=CodeReviewResult,
)
result: CodeReviewResult = response.choices[0].message.parsed

# 方法 3：Anthropic Tool Use 模式
response = client.messages.create(
    model="claude-opus-4-6",
    messages=[{"role": "user", "content": review_prompt}],
    tools=[{
        "name": "submit_review",
        "description": "提交代码审查结果",
        "input_schema": CodeReviewResult.model_json_schema()
    }],
    tool_choice={"type": "tool", "name": "submit_review"}
)

---

三、提示词模板系统

模板引擎设计

from string import Template
from pathlib import Path
import yaml

class PromptTemplate:
    """提示词模板引擎"""

    def __init__(self, template_dir: str = "./prompts"):
        self.template_dir = Path(template_dir)
        self.templates = {}
        self.partials = {}    # 可复用的部分模板
        self._load_all()

    def _load_all(self):
        """加载所有模板文件"""
        for f in self.template_dir.glob("**/*.yaml"):
            name = f.stem
            data = yaml.safe_load(f.read_text(encoding="utf-8"))
            self.templates[name] = data

        for f in (self.template_dir / "partials").glob("*.yaml"):
            name = f.stem
            self.partials[name] = yaml.safe_load(
                f.read_text(encoding="utf-8")
            )

    def render(self, template_name: str, **kwargs) -> str:
        """渲染模板"""
        template_data = self.templates[template_name]

        # 解析 partial 引用
        resolved = self._resolve_partials(template_data)

        # 组装 System Prompt
        parts = []

        if "role" in resolved:
            parts.append(f"你是{resolved['role']}。")

        if "context" in resolved:
            ctx = self._render_section(resolved["context"], kwargs)
            parts.append(f"<context>\n{ctx}\n</context>")

        if "task" in resolved:
            task = self._render_section(resolved["task"], kwargs)
            parts.append(f"<task>\n{task}\n</task>")

        if "constraints" in resolved:
            constraints = "\n".join(
                f"- {c}" for c in resolved["constraints"]
            )
            parts.append(f"<constraints>\n{constraints}\n</constraints>")

        if "output_format" in resolved:
            fmt = resolved["output_format"]
            parts.append(f"<output_format>\n{fmt}\n</output_format>")

        if "examples" in resolved:
            examples = self._render_examples(resolved["examples"], kwargs)
            parts.append(f"<examples>\n{examples}\n</examples>")

        return "\n\n".join(parts)

    def _resolve_partials(self, data: dict) -> dict:
        """解析 partial 引用"""
        resolved = {}
        for key, value in data.items():
            if isinstance(value, str) and value.startswith("$partial:"):
                partial_name = value.split(":")[1]
                resolved[key] = self.partials[partial_name]
            else:
                resolved[key] = value
        return resolved

    def _render_section(self, template_str: str, kwargs: dict) -> str:
        """渲染单个段落"""
        return Template(template_str).safe_substitute(kwargs)

模板文件结构

prompts/
  partials/
    output_json.yaml         # 通用 JSON 输出格式
    output_markdown.yaml     # 通用 Markdown 输出格式
    safety_rules.yaml        # 通用安全规则
    code_style.yaml          # 代码风格约束
  code_review.yaml           # 代码审查模板
  data_analysis.yaml         # 数据分析模板
  document_summary.yaml      # 文档摘要模板
  bug_diagnosis.yaml         # Bug 诊断模板

模板 YAML 示例

# prompts/code_review.yaml
name: code_review
version: "2.1"
description: "代码审查提示词模板"

role: "高级代码审查专家，拥有 10 年以上的软件工程经验"

context: |
  项目名称：$project_name
  技术栈：$tech_stack
  代码规范：$code_standard
  审查范围：$review_scope

task: |
  审查以下代码变更，识别安全漏洞、性能问题、
  可维护性问题和逻辑错误。

constraints:
  - 每个发现必须包含具体的行号和代码片段
  - 严重级别分为 critical/major/minor/suggestion
  - 必须提供具体的修复建议，不能只说"需要优化"
  - 关注实际影响，避免吹毛求疵
  - 如果代码质量良好，明确说明并给出肯定

output_format: $partial:output_json

examples:
  - input: |
      def get_user(id):
          query = f"SELECT * FROM users WHERE id = {id}"
          return db.execute(query)
    output: |
      {
        "findings": [{
          "severity": "critical",
          "category": "security",
          "line": 2,
          "description": "SQL 注入漏洞",
          "suggestion": "使用参数化查询",
          "code": "query = 'SELECT * FROM users WHERE id = %s'\ndb.execute(query, (id,))"
        }]
      }

---

四、提示词组合模式

管道模式（Pipeline）

class PromptPipeline:
    """提示词管道：将复杂任务拆解为多个简单提示词的序列"""

    def __init__(self, stages: list[PromptStage]):
        self.stages = stages

    def execute(self, input_data: dict) -> dict:
        current = input_data
        for stage in self.stages:
            prompt = stage.template.render(**current)
            response = llm.generate(prompt)
            current = {**current, stage.output_key: response}
        return current

# 使用示例：文档分析管道
pipeline = PromptPipeline([
    PromptStage(
        name="extract",
        template=templates["document_extract"],
        output_key="extracted_data"
    ),
    PromptStage(
        name="analyze",
        template=templates["data_analysis"],
        output_key="analysis"
    ),
    PromptStage(
        name="summarize",
        template=templates["executive_summary"],
        output_key="summary"
    ),
])

result = pipeline.execute({"document": document_text})
# result = {document, extracted_data, analysis, summary}

分支模式（Router）

class PromptRouter:
    """提示词路由：根据输入特征选择不同的提示词"""

    def __init__(self, routes: dict[str, PromptTemplate]):
        self.routes = routes
        self.classifier = templates["intent_classifier"]

    def route(self, user_input: str) -> str:
        # 先分类
        classification = llm.generate(
            self.classifier.render(input=user_input)
        )
        intent = json.loads(classification)["intent"]

        # 再路由
        template = self.routes.get(intent)
        if not template:
            template = self.routes["default"]

        return template.render(input=user_input)

router = PromptRouter({
    "code_question": templates["code_expert"],
    "data_question": templates["data_analyst"],
    "general_question": templates["general_assistant"],
    "default": templates["general_assistant"],
})

增强模式（Augmentation）

class PromptAugmenter:
    """提示词增强：动态注入上下文信息"""

    def augment(self, base_prompt: str,
                augmentations: list[Augmentation]) -> str:
        """将检索到的信息注入提示词"""
        parts = [base_prompt]

        for aug in augmentations:
            if aug.type == "rag":
                # RAG 检索增强
                relevant_docs = self.retriever.search(aug.query)
                parts.append(
                    f"<reference_documents>\n"
                    f"{self._format_docs(relevant_docs)}\n"
                    f"</reference_documents>"
                )

            elif aug.type == "memory":
                # 记忆增强
                memories = self.memory.recall(aug.query)
                parts.append(
                    f"<relevant_memory>\n"
                    f"{self._format_memories(memories)}\n"
                    f"</relevant_memory>"
                )

            elif aug.type == "tool_results":
                # 工具结果注入
                parts.append(
                    f"<tool_output tool='{aug.tool_name}'>\n"
                    f"{aug.result}\n"
                    f"</tool_output>"
                )

        return "\n\n".join(parts)

---

五、Few-Shot 示例工程

示例选择策略

class ExampleSelector:
    """动态示例选择器"""

    def __init__(self, example_store: list[dict]):
        self.examples = example_store
        self.embeddings = self._build_embeddings()

    def select(self, query: str, k: int = 3,
               strategy: str = "semantic") -> list[dict]:
        """选择最相关的示例"""
        if strategy == "semantic":
            return self._semantic_select(query, k)
        elif strategy == "diverse":
            return self._diverse_select(query, k)
        elif strategy == "difficulty_matched":
            return self._difficulty_matched(query, k)

    def _semantic_select(self, query: str, k: int) -> list[dict]:
        """语义相似度选择"""
        query_embedding = embed(query)
        similarities = [
            (ex, cosine_similarity(query_embedding, emb))
            for ex, emb in zip(self.examples, self.embeddings)
        ]
        similarities.sort(key=lambda x: x[1], reverse=True)
        return [ex for ex, _ in similarities[:k]]

    def _diverse_select(self, query: str, k: int) -> list[dict]:
        """多样性选择：确保示例覆盖不同的模式"""
        # 先选最相关的
        candidates = self._semantic_select(query, k * 3)
        # 再用 MMR（最大边际相关性）去重
        selected = []
        for candidate in candidates:
            if len(selected) >= k:
                break
            if not self._too_similar_to_selected(candidate, selected):
                selected.append(candidate)
        return selected

示例格式化

def format_examples(examples: list[dict],
                    style: str = "xml") -> str:
    """将示例格式化为提示词片段"""
    if style == "xml":
        parts = []
        for i, ex in enumerate(examples, 1):
            parts.append(f"""<example_{i}>
<input>{ex['input']}</input>
<thinking>{ex.get('thinking', '')}</thinking>
<output>{ex['output']}</output>
</example_{i}>""")
        return "\n\n".join(parts)

    elif style == "markdown":
        parts = []
        for i, ex in enumerate(examples, 1):
            parts.append(f"""### Example {i}

**Input:** {ex['input']}

**Output:** {ex['output']}""")
        return "\n\n".join(parts)

---

六、提示词工程化清单

设计阶段

提示词设计清单：
- [ ] 明确定义了角色（role）
- [ ] 提供了充分的上下文（context）
- [ ] 任务描述具体、无歧义
- [ ] 约束条件完整（做什么、不做什么）
- [ ] 输出格式有明确的 Schema
- [ ] 提供了 2-3 个高质量示例
- [ ] 考虑了边界情况的处理
- [ ] 考虑了安全约束（不输出敏感信息等）

工程化阶段

提示词工程化清单：
- [ ] 使用模板引擎管理，不在代码中硬编码
- [ ] 模板有版本号和变更记录
- [ ] 变量用 $variable 或 {variable} 标记
- [ ] 可复用部分抽取为 partial
- [ ] 有自动化测试覆盖
- [ ] 有回归检测机制
- [ ] 模板文件纳入版本控制
- [ ] 有 Token 消耗估算

---

参考资料

• Anthropic Prompt Engineering 官方指南
• OpenAI Prompt Engineering 最佳实践
• LangChain Prompt Templates 文档
• LMQL：面向 LLM 的查询语言
• DSPy：Programming with Foundation Models

---

Maurice | maurice_wen@proton.me