YAML格式Prompt工程：读取、渲染与实战使用完全指南

---

一、为什么用YAML管理Prompt？

在工程化实践中，将Prompt从"硬编码字符串"升级为YAML配置文件，带来五个核心优势：

特性	说明	适用场景
结构化	分离系统提示、变量、配置参数	团队协作、版本管理
动态化	Jinja2模板支持条件判断与循环	个性化内容生成
可组合	继承与多态机制	多轮对话、Agent系统
可追踪	变量与元数据完整记录	A/B测试、效果分析
安全	环境变量隔离敏感信息	生产环境部署

---

二、标准YAML Prompt结构

一个完整的YAML Prompt文件由五个层次组成：元信息、系统提示、变量定义、用户模板、模型配置。

# prompt.yaml
name: "代码审查助手"
description: "专业的代码质量和安全性审查专家"
version: "1.0.0"

# ① 系统提示
system_prompt: |
  你是一位资深的代码审查专家，拥有10年以上的软件开发经验。
  你的职责是：
  1. 识别代码中的潜在bug和逻辑错误
  2. 检查安全漏洞（SQL注入、XSS等）
  3. 评估代码性能和可维护性
  4. 提供具体的改进建议

# ② 动态变量定义（支持类型校验与默认值）
variables:
  - name: "code_language"
    type: "string"
    required: true
    description: "编程语言"
    default: "Python"

  - name: "review_focus"
    type: "array"
    required: false
    description: "审查重点"
    default: ["security", "performance"]
    options: ["security", "performance", "readability", "architecture"]

  - name: "code_snippet"
    type: "text"
    required: true
    description: "待审查的代码"

# ③ 用户消息模板（Jinja2语法）
user_template: |
  请审查以下{{ code_language }}代码：

  ```{{ code_language }}
  {{ code_snippet }}

审查重点：{{ review_focus | join(‘、’) }}

请按以下格式输出：

• 🔴 严重问题
• 🟡 建议改进
• 🟢 良好实践

④ 模型参数配置

settings:
temperature: 0.2
max_tokens: 2000
top_p: 0.95
response_format: “markdown”

---

## 三、基础读取与渲染流程

### 步骤一：安装依赖

```bash
pip install pyyaml jinja2

步骤二：读取YAML文件

import yaml
from jinja2 import Template

def load_prompt_yaml(file_path):
    with open(file_path, 'r', encoding='utf-8') as f:
        return yaml.safe_load(f)

config = load_prompt_yaml('prompt.yaml')
print(f"Prompt名称: {config['name']}")

步骤三：变量渲染（核心步骤）

渲染逻辑分三步：合并默认值 → Jinja2渲染模板 → 组装消息列表。

def render_prompt(config, user_inputs):
    # 1. 合并默认值与用户输入
    final_values = {}
    for var in config.get('variables', []):
        name = var['name']
        if name in user_inputs:
            final_values[name] = user_inputs[name]
        elif 'default' in var:
            final_values[name] = var['default']
        elif var.get('required', False):
            raise ValueError(f"缺少必需变量: {name}")

    # 2. 渲染用户模板
    rendered_user = Template(config['user_template']).render(**final_values)

    # 3. 组装完整消息
    return {
        "messages": [
            {"role": "system", "content": config['system_prompt']},
            {"role": "user",   "content": rendered_user}
        ],
        "settings": config.get('settings', {}),
        "variables_used": final_values
    }

# 调用示例
result = render_prompt(config, {
    "code_language": "Python",
    "code_snippet": "def login(user, pwd):\n    query = f\"SELECT * FROM users WHERE name='{user}'\"",
    "review_focus": ["security", "performance"]
})
print(result['messages'][1]['content'])

---

四、四大实战场景

场景一：多轮对话管理（客服机器人）

设计思路：在YAML中定义每一轮的触发条件和模板，Python负责状态维护与条件匹配，实现轮次驱动的对话流。

# customer_service.yaml
name: "智能客服助手"
type: "multi_turn"

system_prompt: |
  你是电商平台的智能客服，语气亲切专业。
  当前对话状态: {{ conversation_state | tojson }}

turns:
  - turn: 1
    condition: "intent == 'inquiry'"
    prompt: "您好！请问有什么可以帮您？您可以查询订单、申请退款或咨询商品。"

  - turn: 2
    condition: "intent == 'order_query' and not order_id"
    prompt: "请提供您的订单号（格式：ORD-XXXXXX），我立即为您查询。"
    extract:
      - variable: "order_id"
        pattern: "ORD-[A-Z0-9]{6}"

  - turn: 3
    condition: "order_id and not issue_resolved"
    prompt: |
      订单 {{ order_id }} 状态：{{ order_status }}
      {% if order_status == 'shipped' %}
      预计送达：{{ estimated_delivery }}
      {% endif %}
      是否解决了您的问题？

import re, yaml
from jinja2 import Template

class MultiTurnPromptManager:
    def __init__(self, yaml_path):
        self.config = yaml.safe_load(open(yaml_path))
        self.state = {"current_intent": None, "turn_count": 0, "extracted_vars": {}}

    def get_next_prompt(self, intent):
        self.state["turn_count"] += 1
        self.state["current_intent"] = intent
        for turn in self.config['turns']:
            if self._check_condition(turn['condition']):
                return Template(turn['prompt']).render(
                    conversation_state=self.state,
                    **self.state['extracted_vars']
                )
        return "抱歉，我不太理解，请重新描述您的问题。"

    def extract_variables(self, user_message, turn_index):
        turn = self.config['turns'][turn_index - 1]
        for rule in turn.get('extract', []):
            match = re.search(rule['pattern'], user_message)
            if match:
                self.state['extracted_vars'][rule['variable']] = match.group(0)

---

场景二：条件分支（A/B测试Prompt）

设计思路：在YAML中声明多个分支及其触发条件，路由器按顺序匹配第一个满足条件的分支，实现策略与内容解耦。

# ab_test_prompt.yaml
name: "营销文案生成器"

branches:
  - id: "version_a_emotional"
    condition: "user_segment == 'young' and channel == 'social'"
    system_prompt: |
      你是一位懂Z世代的营销专家，使用网络热梗和emoji。
      语气：活泼、共情、短句为主。
    user_template: |
      为{{ product_name }}写一条{{ platform }}文案，目标用户是{{ age }}岁的{{ interest }}爱好者。
      要求：带话题标签，使用当下流行语，不超过100字。

  - id: "version_b_professional"
    condition: "user_segment == 'professional' or channel == 'email'"
    system_prompt: |
      你是一位B2B营销专家，强调数据支撑和ROI。
      语气：专业、权威、结构化。
    user_template: |
      撰写一封关于{{ product_name }}的营销邮件，面向{{ job_title }}。
      必须包含：痛点分析、解决方案、客户案例、CTA按钮文案。

  - id: "version_c_urgency"
    condition: "campaign_type == 'flash_sale'"
    system_prompt: |
      你是促销文案高手，制造稀缺感和紧迫感。
    user_template: |
      倒计时{{ hours_left }}小时！{{ product_name }}限时{{ discount }}折！
      强调：库存紧张、时间限制、立即行动。

from jinja2 import Environment, BaseLoader

class ConditionalPromptRouter:
    def __init__(self, yaml_path):
        with open(yaml_path, 'r') as f:
            self.config = yaml.safe_load(f)
        self.env = Environment(loader=BaseLoader())

    def evaluate_condition(self, condition_str, context):
        template = self.env.from_string("{{ " + condition_str + " }}")
        try:
            return template.render(**context).lower() in ['true', '1', 'yes']
        except:
            return False

    def route_and_render(self, routing_context, content_variables):
        branch = next(
            (b for b in self.config['branches']
             if self.evaluate_condition(b['condition'], routing_context)),
            None
        )
        if not branch:
            raise ValueError("没有匹配的条件分支")

        return {
            "branch_id": branch['id'],
            "messages": [
                {"role": "system", "content": self.env.from_string(branch['system_prompt']).render(**content_variables)},
                {"role": "user",   "content": self.env.from_string(branch['user_template']).render(**content_variables)}
            ]
        }

# 调用示例：年轻用户 + 社交媒体渠道
router = ConditionalPromptRouter('ab_test_prompt.yaml')
result = router.route_and_render(
    routing_context={"user_segment": "young", "channel": "social", "campaign_type": "regular"},
    content_variables={"product_name": "降噪耳机Pro", "platform": "小红书", "age": 22, "interest": "音乐"}
)
print(f"命中分支: {result['branch_id']}")

---

场景三：动态Few-Shot示例管理

设计思路：在YAML中维护示例库，运行时根据相似度（或随机策略）动态选取最相关的示例注入模板，避免无效的固定示例。

# few_shot_prompt.yaml
name: "情感分析专家"

system_prompt: |
  你是情感分析专家，将文本分类为：正面、负面、中性。
  严格遵循示例中的格式和推理逻辑。

example_pool:
  - id: 1
    category: "product_review"
    examples:
      - input: "这款手机拍照效果惊艳，夜景模式特别清晰！"
        output: "正面 | 理由：明确表达对拍照功能的满意，使用积极词汇'惊艳'、'清晰'"
      - input: "物流太慢了，等了一周才到货，包装还破损了。"
        output: "负面 | 理由：抱怨配送速度和包装质量，情绪负面"
  - id: 2
    category: "service_feedback"
    examples:
      - input: "客服响应很快，但解决方案不太实用。"
        output: "中性 | 理由：同时包含正面（响应快）和负面（方案不实用）要素，整体平衡"

selection:
  strategy: "similarity"   # 可选：random / category_match
  top_k: 3

prompt_template: |
  {% if examples %}
  以下是一些参考示例：
  {% for ex in examples %}
  文本：{{ ex.input }}
  分析：{{ ex.output }}
  ---
  {% endfor %}
  {% endif %}

  请分析以下文本：{{ target_text }}
  要求：先给出分类，再简要说明理由（20字内）。

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

class DynamicFewShotPrompt:
    def __init__(self, yaml_path, embedding_fn=None):
        with open(yaml_path, 'r') as f:
            self.config = yaml.safe_load(f)
        self.embedding_fn = embedding_fn or (lambda t: np.random.randn(1536))
        self.examples = [
            {**ex, "category": pool['category']}
            for pool in self.config['example_pool']
            for ex in pool['examples']
        ]

    def select_examples(self, target_text, top_k=3):
        target_vec = self.embedding_fn(target_text)
        scored = sorted(
            self.examples,
            key=lambda ex: cosine_similarity(
                [target_vec], [self.embedding_fn(ex['input'])]
            )[0][0],
            reverse=True
        )
        return scored[:top_k]

    def build_prompt(self, target_text):
        selected = self.select_examples(target_text, self.config['selection']['top_k'])
        user_content = Template(self.config['prompt_template']).render(
            examples=selected, target_text=target_text
        )
        return {
            "messages": [
                {"role": "system", "content": self.config['system_prompt']},
                {"role": "user",   "content": user_content}
            ],
            "selected_examples": [ex['input'][:30] + "..." for ex in selected]
        }

# 调用示例
few_shot = DynamicFewShotPrompt('few_shot_prompt.yaml')
result = few_shot.build_prompt("这家餐厅的牛排煎得恰到好处，服务也很周到。")
print(result['messages'][1]['content'])

---

场景四：工具调用（Function Calling Agent）

设计思路：在YAML中声明工具清单及参数规范，Agent负责格式转换、参数校验和工具执行，实现工具配置与调用逻辑分离。

# agent_tools.yaml
name: "数据分析Agent"

system_prompt: |
  你是数据分析专家，可调用工具查询数据库或生成图表。
  当需要数据时，必须调用工具，不得编造数据。

available_tools:
  - name: "query_database"
    description: "执行SQL查询"
    parameters:
      type: "object"
      properties:
        sql:
          type: "string"
          description: "标准SQL语句"
        limit:
          type: "integer"
          default: 100
      required: ["sql"]

  - name: "generate_chart"
    description: "生成数据可视化图表"
    parameters:
      type: "object"
      properties:
        chart_type:
          type: "string"
          enum: ["line", "bar", "pie", "scatter"]
        data_source:
          type: "string"
        x_axis:
          type: "string"
        y_axis:
          type: "string"
      required: ["chart_type", "data_source"]

  - name: "send_email"
    description: "发送分析报告邮件"
    parameters:
      type: "object"
      properties:
        to:
          type: "string"
        subject:
          type: "string"
        attachment:
          type: "string"
      required: ["to", "subject"]

import json, yaml
from typing import Dict, Any

class ToolCallingAgent:
    def __init__(self, yaml_path):
        with open(yaml_path, 'r') as f:
            self.config = yaml.safe_load(f)
        self.tools = {t['name']: t for t in self.config['available_tools']}

    def get_tools_for_llm(self):
        """转换为 OpenAI/Claude 标准工具格式"""
        return [
            {"type": "function", "function": {
                "name": name,
                "description": tool['description'],
                "parameters": tool['parameters']
            }}
            for name, tool in self.tools.items()
        ]

    def execute_tool(self, tool_name: str, parameters: Dict) -> Any:
        """执行工具（此处为模拟实现，生产环境替换为真实逻辑）"""
        mock_results = {
            "query_database": {"status": "success", "rows": 150,
                               "preview": [{"date": "2024-01", "sales": 10000}]},
            "generate_chart":  {"status": "success", "chart_url": "/charts/bar_123.png"},
            "send_email":      {"status": "success", "message_id": "msg_456"}
        }
        return mock_results.get(tool_name, {"error": "Unknown tool"})

    def process_llm_response(self, llm_response: Dict) -> Dict:
        tool_calls = llm_response.get('message', {}).get('tool_calls', [])
        if not tool_calls:
            return {"type": "final_answer", "content": llm_response['message'].get('content', '')}

        results = []
        for call in tool_calls:
            name = call['function']['name']
            args = json.loads(call['function']['arguments'])
            # 参数校验
            required = self.tools[name]['parameters'].get('required', [])
            missing = [p for p in required if p not in args]
            if missing:
                results.append({"tool": name, "error": f"缺少必需参数: {missing}"})
                continue
            results.append({"tool": name, "arguments": args, "result": self.execute_tool(name, args)})

        return {"type": "tool_results", "results": results, "follow_up_required": True}

# 调用流程示例
agent = ToolCallingAgent('agent_tools.yaml')

mock_llm_response = {
    "message": {"tool_calls": [{
        "id": "call_1",
        "function": {
            "name": "query_database",
            "arguments": json.dumps({
                "sql": "SELECT region, SUM(sales) FROM sales WHERE month='2024-02' GROUP BY region",
                "limit": 50
            })
        }
    }]}
}

result = agent.process_llm_response(mock_llm_response)
print(json.dumps(result, indent=2, ensure_ascii=False))

---

五、高级技巧

技巧一：环境变量注入（敏感信息隔离）

通过自定义 !ENV 标签，将API Key等敏感信息从YAML文件中剥离，只保留占位符。

# config.yaml
api_config:
  base_url: !ENV ${OPENAI_API_BASE}
  api_key:  !ENV ${OPENAI_API_KEY}
  model:    !ENV ${MODEL_NAME:gpt-4}   # 冒号后为默认值

import os, yaml

def env_constructor(loader, node):
    value = loader.construct_scalar(node)
    inner = value[2:-1]  # 去掉 ${ 和 }
    if ':' in inner:
        var_name, default = inner.split(':', 1)
        return os.getenv(var_name, default)
    return os.getenv(inner)

yaml.SafeLoader.add_constructor('!ENV', env_constructor)
config = yaml.safe_load(open('config.yaml'))

---

技巧二：Prompt版本注册与流量路由

通过注册表统一管理多版本Prompt，支持灰度发布和A/B流量分配。

# prompt_registry.yaml
prompts:
  - name: "summary_v1"
    version: "1.0.0"
    status: "deprecated"
    file: "./prompts/summary_v1.yaml"

  - name: "summary_v2"
    version: "2.0.0"
    status: "active"
    file: "./prompts/summary_v2.yaml"
    traffic_percentage: 80

  - name: "summary_v2_experimental"
    version: "2.1.0-beta"
    status: "experiment"
    file: "./prompts/summary_v2_beta.yaml"
    traffic_percentage: 20
    target_users: "beta_group"

class PromptPipeline:
    def __init__(self, registry_path: str):
        self.registry = yaml.safe_load(open(registry_path))
        self.cache = {}

    def load(self, prompt_name: str, context: dict = None) -> dict:
        # 1. 查找注册表
        meta = next((p for p in self.registry['prompts'] if p['name'] == prompt_name), None)
        if not meta:
            raise ValueError(f"Prompt '{prompt_name}' 不存在")

        # 2. 加载并缓存
        if prompt_name not in self.cache:
            self.cache[prompt_name] = yaml.safe_load(open(meta['file']))
        config = self.cache[prompt_name]

        # 3. 验证变量
        ctx = context or {}
        for var in config.get('variables', []):
            if var.get('required') and var['name'] not in ctx and 'default' not in var:
                raise ValueError(f"缺少必需变量: {var['name']}")

        # 4. 渲染并返回
        rendered = self._render(config, ctx)
        return {
            "messages": rendered['messages'],
            "settings": config.get('settings', {}),
            "metadata": {"version": meta['version'], "variables_used": list(ctx.keys())}
        }

    def _render(self, config, context):
        user_content = Template(config['user_template']).render(**context)
        return {
            "messages": [
                {"role": "system", "content": config['system_prompt']},
                {"role": "user",   "content": user_content}
            ]
        }

# 使用
pipeline = PromptPipeline('prompt_registry.yaml')
result = pipeline.load('summary_v2', {'text': '需要总结的内容...', 'style': 'bullet_points'})

---

六、总结

YAML Prompt工程的本质是将提示工程从字符串拼接升级为可配置、可版本化、可监控的系统工程。核心设计原则可归纳为三点：

声明与逻辑分离：YAML负责"写什么"，Python负责"怎么用"，两者职责清晰，互不耦合。

渐进式复杂度：从基础模板起步，按需引入条件分支、Few-Shot、工具调用等能力，避免过度设计。

可观测性优先：每次渲染都记录变量使用情况、命中分支、版本信息，为后续优化提供数据基础。