林清扬的博客

目标 路径 时间 结果 信息 定位 闭环 复盘 精力 极限 稳态 框架

【AI Agent 知识库】05-Prompt工程与约束输出-详解版

account_circle 林清杨 schedulefolder_open2026, AI, AI Agent 知识库
内容纲要

模块五：Prompt 工程与约束输出（详解版）

覆盖：模板管理、Few-shot、输出约束、JSON Mode、Pydantic 验证

目录

必须掌握的概念

5.1 Prompt Template（提示模板）

定义：
预定义的提示模板，通过填充动态内容生成完整提示。

优势：

  • 可复用
  • 易维护
  • 版本控制

类型：

类型 说明 示例
System Prompt 系统级提示，定义角色 "你是一个专业的Python开发助手"
User Prompt 用户输入 用户的具体问题
Chat History 对话历史 历史对话记录
Few-shot 少样本示例 提供示例引导
Instruction 指令说明 "请用JSON格式回答"

5.2 Few-shot Learning（少样本学习）

定义：
在 Prompt 中提供示例，引导 LLM 学习期望的行为。

作用：

  • 提升准确度
  • 规范输出格式
  • 减少幻觉

示例：

# 没有 Few-shot
prompt = "将 '2024-01-01' 转换为 '1月1日'"
# 可能输出：January 1, 2024（格式不确定）

# 有 Few-shot
prompt = """
将日期转换为中文格式

示例1：
输入：2024-01-01
输出：2024年1月1日

示例2：
输入：2024-12-25
输出：2024年12月25日

现在请转换：2024-06-15
"""
# 更可能输出：2024年6月15日（格式一致）

5.3 Output Format（输出格式约束）

格式类型：

格式 适用场景 约束度
JSON 结构化数据 中
XML 数据交换 低
YAML 配置文件 中
Markdown 文档生成 低
CSV 表格数据 中

5.4 JSON Mode

定义：
OpenAI 的 response_format={"type": "json_object"} 功能，强制输出合法 JSON。

优势：

  • 保证输出格式正确
  • 减少 parse 错 fancys
  • 提高稳定性

代码示例：

response = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "请用JSON回答"}],
    response_format={"type": "json_object"}  # 强制 JSON
)

# 输出保证是合法 JSON
result = json.loads(response.choices[0].message.content)

5.5 Pydantic 验证

定义：
使用 Pydantic 库定义数据模型，对 LLM 输出进行类型验证。

优势：

  • 类型安全
  • 自动类型转换
  • 详细的错误信息
  • 可用于 JSON Schema 生成

代码示例：

from pydantic import BaseModel, Field, validator

class Answer(BaseModel):
    reasoning: str = Field(description="推理过程")
    answer: str = Field(description="最终答案")
    confidence: float = Field(
        description="置信度 [0-1]",
        ge=0, le=1
    )

# 验证 LLM 输出
try:
    answer = Answer.model_validate_json(llm_output)
except ValidationError as e:
    print(f"输出格式错误: {e}")

关键设计点

5.1 Prompt Template 管理

# prompt/template_manager.py
"""
Prompt 模板管理器
包含：模板加载、变量填充、版本管理
"""

from typing import Dict, List, Optional
from dataclasses import dataclass
from string import Template
import json
import yaml

@dataclass
class PromptTemplate:
    name: str
    content: str
    variables: List[str]
    version: str = "1.0"
    metadata: Dict = None

class PromptManager:
    def __init__(self):
        self._templates: Dict[str, PromptTemplate] = {}

    def load_from_file(self, file_path: str):
        """从文件加载模板"""
        if file_path.endswith('.json'):
            return self._load_from_json(file_path)
        elif file_path.endswith(('.yaml', '.yml')):
            return self._load_from_yaml(file_path)
        elif file_path.endswith('.md'):
            return self._load_from_markdown(file_path)
        else:
            raise ValueError(f"不支持的格式: {file_path}")

    def _load_from_json(self, file_path: str) -> PromptTemplate:
        """从 JSON 加载"""
        with open(file_path, 'r', encoding='utf-8') as f:
            data = json.load(f)

        template = PromptTemplate(
            name=data['name'],
            content=data['content'],
            variables=data.get('variables', []),
            version=data.get('version', '1.0'),
            metadata=data.get('metadata')
        )

        self._templates[template.name] = template
        return template

    def _load_from_yaml(self, file_path: str) -> PromptTemplateverload(self, file_path) -> PromptTemplate:
        """从 Markdown 加载"""
        with open(file_path, 'r', encoding='utf-8') as f:
            content = f.read()

        # 提取元数据（从 YAML front matter）
        template = PromptTemplate(
            name=file_path.split('/')[-1],
            content=content,
            variables=[],
            version="1.0",
            metadata=None
        )

        self._templates[template.name] = template
        return template

    def fill_template(
        self,
        template_name: str,
        variables: Dict
    ) -> str:
        """填充模板变量"""
        template = self._templates.get(template_name)
        if template is None:
            raise ValueError(f"模板不存在: {template_name}")

        # 检查所有必需变量
        missing_vars = set(template.variables) - set(variables.keys())
        if missing_vars:
            raise ValueError(f"缺少变量: {missing_vars}")

        # 使用字符串模板
        string_template = Template(template.content)
        return string_template.safe_substitute(**variables)

    def get_template(self, name: str) -> Optional[PromptTemplate]:
        """获取模板"""
        return self._templates.get(name)

    def list_templates(self) -> List[str]:
        """列出所有模板"""
        return list(self._templates.keys())

# ============ 模板示例 ============

# system_prompt.json
{
    "name": "system_prompt",
    "content": "你是一个{{role}}，擅长{{expertise}}。\n\n请遵循以下规则：\n- {{rule1}}\n- {{rule2}}",
    "variables": ["role", "expertise", "rule1", "rule2"],
    "version": "1.0",
    "metadata": {
        "category": "system",
        "author": "AI Team"
    }
}

# user_prompt.md
---
name: user_prompt
variables:
  - task
  - context
---

你是一个智能助手，请帮助用户完成以下任务：

任务：{{task}}

上下文信息：
{{context}}

请提供详细的解决方案。

# ============ 使用示例 ============

manager = PromptManager()

# 加载模板
manager.load_from_file("system_prompt.json")
manager.load_from_file("user_prompt.md")

# 填充模板
system_prompt = manager.fill_template(
    "system_prompt",
    variables={
        "role": "Python开发专家",
        "expertise": "Python并发编程和性能优化",
        "rule1": "代码简洁易读",
        "rule2": "注重性能和最佳实践"
    }
)

user_prompt = manager.fill_template(
    "user_prompt",
    variables={
        "task": "编写一个高并发的计数器",
        "context": "需要支持多线程，保证线程安全，性能优于synchronized"
    }
)

print(system_prompt)
print(user_prompt)

5.2 Few-shot 管理

# prompt/fewshot_manager.py
"""
Few-shot 管理器
包含：示例管理、动态添加、相似度检索
"""

from typing import List, Dict, Optional, Callable
from dataclasses import dataclass
import numpy as np
from sentence_transformers import SentenceTransformer

@dataclass
class Example:
    input: str
    output: str
    category: Optional[str] = None
    metadata: Dict = None

class FewShotManager:
    def __init__(
        self,
        embedder: Optional[SentenceTransformer] = None,
        max_examples: int = 10
    ):
        self.embedder = embedder
        self.max_examples = max_examples
        self._examples: Dict[str, List[Example]] = {}

    def add_examples(
        self,
        category: str,
        examples: List[Example]
    ):
        """添加示例到类别"""
        if category not in self._examples:
            self._examples[category] = []

        self._examples[category].extend(examples)

    def select_examples(
        self,
        query: str,
        category: Optional[str] = None,
        k: int = 5
    ) -> List[Example]:
        """选择最相关的示例"""
        if category:
            examples = self._examples.get(category, [])[:k]
            return examples

        # 基于相似度选择
        if self.embedder:
            return self._select_by_similarity(query, k)
        else:
            return self._select_random(k)

    def _select_by_similarity(self, query: str, k: int) -> List[Example]:
        """基于相似度选择"""
        query_embedding = self.embedder.encode(query)
        similarities = []

        for category, examples in self._examples.items():
            for example in examples:
                example_embedding = self.embedder.encode(example.input)
                similarity = np.dot(query_embedding, example_embedding) / (
                    np.linalg.norm(query_embedding) * np.linalg.norm(example_embedding)
                )
                similarities.append((similarity, example))

        # 按相似度排序
        similarities.sort(key=lambda x: x[0], reverse=True)

        # 返回 top-k
        return [example for _, example in similarities[:k]]

    def _select_random(self, k: int) -> List[Example]:
        """随机选择"""
        import random

        all_examples = []
        for examples in self._examples.values():
            all_examples.extend(examples)

        if len(all_examples) <= k:
            return all_examples

        return random.sample(all_examples, k)

    def format_examples(self, examples: List[Example]) -> str:
        """格式化示例"""
        formatted = []
        for i, example in enumerate(examples, 1):
            formatted.append(f"示例 {i}:")
            formatted.append(f"输入：{example.input}")
            formatted.append(f"输出：{example.output}")
            if example.category:
                formatted.append(f"类别：{example.category}")
            formatted.append("")

        return "\n".join(formatted)

# ============ 使用示例 ============

# 创建 Few-shot 管理器
fewshot_manager = FewShotManager(
    embedder=SentenceTransformer('BAAI/bge-small-zh-v1.5'),
    max_examples=10
)

# 添加分类示例
fewshot_manager.add_examples("code_generation", [
    Example(
        input="计算斐波那契数列",
        output="```python\ndef fib(n):\n    if n <= 1:\n        return n\n    a, b = 0, 1\n    for _ in range(2, n + 1):\n        a, b = b, a + b\n    return b\n```"
    ),
    Example(
        input="计算阶乘",
        output="```python\nimport math\n\ndef factorial(n):\n    return math.prod(range(1, n + 1))\n```"
    )
])

fewshot_manager.add_examples("data_analysis", [
    Example(
        input="计算平均值",
        output="```python\ndef average(numbers):\n    return sum(numbers) / len(numbers)\n```"
    ),
    Example(
        input="计算中位数",
        output="```python\ndef median(numbers):\n    sorted_nums = sorted(numbers)\n    n = len(sorted_nums)\n    return (sorted_nums[n//2] + sorted_nums[(n-1)//2]) / 2\n```"
    )
])

# 选择示例
query = "编写一个快速排序函数"
examples = fewshot_manager.select_examples(query, category="code_generation", k=2)

# 格式化到 Prompt
prompt = f"""
你是一个编程助手，请用Python代码解决问题。

以下是一些参考示例：

{fewshot_manager.format_examples(examples)}

现在请解决以下问题：

问题：{query}

请只输出代码，不要包含其他内容。
"""

5.3 输出约束系统

# prompt/constraint_system.py
"""
输出约束系统
包含：JSON Mode、Function Calling、Pydantic 验证、重试机制
"""

from typing import Any, Dict, Optional
from pydantic import BaseModel, Field, ValidationError
import json
import time
from dataclasses import dataclass

# ============ 输出模型定义 ============

class ReasoningAnswer(BaseModel):
    reasoning: str = Field(description="推理过程")
    answer: str = Field(description="最终答案")
    confidence: float = Field(
        description="置信度 [0-1]",
        ge=0, le=1
    )
    sources: List[str] = Field(default_factory=list, description="信息来源")

class CodeAnswer(BaseModel):
    code: str = Field(description="代码内容")
    language: str = Field(description="编程语言")
    explanation: str = Field(default="", description="代码说明")

class StructuredResponse(BaseModel):
    status: str = Field(description="状态: success/error")
    data: Any = Field(description="响应数据")

# ============ 约束管理器 ============

class OutputConstraintManager:
    def __init__(
        self,
        llm,
        max_retries: int = 3,
        timeout: int = 30
    ):
        self.llm = llm
        self.max_retries = max_retries
        self.timeout = timeout

    def get_structured_response(
        self,
        prompt: str,
        response_model: type,
        use_json_mode: bool = True
    ) -> Any:
        """获取结构化响应"""
        for attempt in range(self.max_retries + 1):
            try:
                if attempt == 0:
                    # 第一次使用 JSON Mode
                    return self._call_with_json_mode(prompt, response_model)
                else:
                    # 后续尝试添加更多说明
                    enhanced_prompt = self._enhance_prompt(prompt, attempt)
                    return self._call_with_json_mode(enhanced_prompt, response_model)

            except ValidationError as e:
                print(f"尝试 {attempt + 1} 验证失败: {e}")
                continue

            except Exception as e:
                print(f"尝试 {attempt + 1} 调用失败: {e}")
                continue

        raise Exception(f"所有尝试都失败")

    def _call_with_json_mode(
        self,
        prompt: str,
        response_model: type
    ) -> Any:
        """使用 JSON Mode 调用"""
        response = self.llm.generate(
            prompt=prompt,
            response_format={"type": "json_object"}
        )

        # 验证并解析
        return response_model.model_validate_json(response)

    def _enhance_prompt(self, prompt: str, attempt: int) -> str:
        """增强 Prompt"""
        additional = f"""

重要：请确保输出是合法的 {attempt + 1} 次！
输出必须符合以下 JSON Schema：
{response_model.model_json_schema()}

请严格遵守格式要求。
"""
        return prompt + additional

# ============ 使用示例 ============

# 模拟 LLM
class MockLLM:
    def generate(self, prompt: str, **kwargs):
        if "JSON" in prompt:
            # 模拟 JSON 输出
            return {
                "reasoning": "这是一个计算示例",
                "answer": "斐波那契数列是 0, 1, 1, 2, 3, 5, 8...",
                "confidence": 0.95,
                "sources": ["数学课本"]
            }
        else:
            return "这是一个回答"

# 创建约束管理器
manager = OutputConstraintManager(
    llm=MockLLM(),
    max_retries=3
)

# 调用
prompt = "请计算斐波那契数列的前10项，并给出推理过程"
result = manager.get_structured_response(
    prompt=prompt,
    response_model=ReasoningAnswer,
    use_json_mode=True
)

print("=" * 60)
print("最终答案：")
print("=" * 60)
print(f"推理过程: {result.reasoning}")
print(f"答案: {result.answer}")
print(f"置信度: {result.confidence}")
print(f"来源: {result.sources}")

常见坑与解决方案

5.1 输出格式不稳定

问题：

  • LLM 输出格式不一致
  • 有时 JSON，有时文本
  • 缺少必需字段

解决方案：

# 使用 JSON Mode + 重试
response = client.chat.completions.create(
    model="gpt-4",
    messages=[...],
    response_format={"type": "json_object"}  # 强制 JSON
)

5.2 字符截断导致格式错误

问题：

  • JSON 在 token 边界被截断
  • {"key": "value （缺少 }）

解决方案：

# 方法 1: 使用 JSON Mode
response_format={"type": "json_object"}

# 方法 2: 手动补全
def complete_json(json_str: str) -> str:
    """补全 JSON"""
    # 计算花括号
    open_braces = json_str.count('{')
    close_braces = json_str.count('}')

    # 补全缺失的括号
    return json_str + '}' * (open_braces - close_braces)

# 方法 3: 重试更长的输出
def ensure_complete_json(llm, prompt, max_retries=3):
    for i in range(max_retries):
        response = llm.generate(prompt)
        try:
            json.loads(response)
            return response
        except:
            # 要求完整输出
            prompt = f"{prompt}\n\n重要：请确保输出完整的JSON，不要截断。"

    raise Exception("无法获取完整 JSON")

面试高频问法

Q1: 如何保证 LLM 输出符合预期格式？

标准回答：

约束方法优先级：

1. JSON Mode（最高优先级）
   - OpenAI: response_format={"type": "json_object"}
   - 保证输出是合法 JSON
   - 无需 parse 和重试

2. Function Calling
   - 定义输出 Schema
   - LLM 自动调用函数
   - 参数保证格式

3. Pydantic 验证
   - 定义输出模型
   - 类型检查 + 范围验证
   - 详细的错误信息

4. Few-shot 示例
   - 提供正确输出示例
   - LLM 学习格式

5. System Prompt �束
   - 明确输出要求
   - 指定格式说明

6. 后处理校验
   - 校验必需字段
   - 校验数据类型
   - 校验取值范围

7. 重试机制
   - 验证失败自动重试
   - 增强 Prompt 重新调用

工程实践：
- 首选：JSON Mode 或 Function Calling
- 次选：Pydantic 验证
-   - 准备重试机制（3 次）

记忆要点

输出约束口诀：

JSON Mode 强制格式
Function Calling 自动验证
Pydantic 类型安全
Few-shot 引导格式
System Prompt 明确要求
重试机制容错

Checklist:
□ 使用 JSON Mode 或 Function Calling
□ 定义 Pydantic 模型
□ 验证失败重试
□ Few-shot 提供示例
□ Prompt 明确格式要求
□ 必需字段检查

文档版本: 1.0

close
arrow_upward