第03课：Prompt Engineering

✍️ Prompt Engineering：与AI对话的艺术

Prompt Engineering是设计和优化与LLM交互的提示文本的技术。对Agent而言，好的Prompt意味着更准确的推理、更可靠的工具调用、更少的幻觉。

📖 核心原则

📊 Prompt技术全景

Prompt Engineering 技术树
├── 基础技术
│   ├── Zero-shot Prompting（零样本提示）
│   ├── Few-shot Prompting（少样本提示）
│   └── Chain-of-Thought / CoT（思维链）
├── 高级技术
│   ├── Self-Consistency（自一致性）
│   ├── Tree-of-Thought / ToT（思维树）
│   ├── ReAct (Reasoning+Acting)
│   └── Reflexion（反思）
├── Agent专用
│   ├── Tool-use Prompting（工具调用提示）
│   ├── Plan-and-Solve（规划与解决）
│   ├── Role Prompting（角色提示）
│   └── System Prompt设计
└── 优化技术
    ├── Prompt Compression（压缩）
    ├── Automatic Prompt Engineer (APE)
    └── DSPy (编译优化)

💻 代码实现：Prompt模板引擎

# Prompt模板引擎与System Prompt构建器
import re, json

class PromptTemplate:
    # Prompt模板 - 支持{{变量}}语法
    def __init__(self, template):
        self.template = template
        self.variables = re.findall(r'\{\{(\w+)\}\}', template)
    
    def render(self, **kwargs):
        result = self.template
        for key, value in kwargs.items():
            result = result.replace("{{" + key + "}}", str(value))
        unfilled = re.findall(r'\{\{(\w+)\}\}', result)
        if unfilled:
            raise ValueError(f"未填充的变量: {unfilled}")
        return result

class SystemPromptBuilder:
    # System Prompt构建器 - 链式调用
    def __init__(self):
        self.role = ""
        self.capabilities = []
        self.constraints = []
        self.output_format = ""
        self.examples = []
        self.safety_rules = []
    
    def set_role(self, role, expertise=""):
        self.role = role + (f"，专长于{expertise}" if expertise else "")
        return self
    def add_capability(self, cap):
        self.capabilities.append(cap); return self
    def add_constraint(self, con):
        self.constraints.append(con); return self
    def set_output_format(self, fmt):
        self.output_format = fmt; return self
    def add_example(self, inp, out):
        self.examples.append({"input": inp, "output": out}); return self
    def add_safety_rule(self, rule):
        self.safety_rules.append(rule); return self
    
    def build(self):
        parts = []
        if self.role:
            parts.append(f"## 角色\n你是{self.role}。")
        if self.capabilities:
            parts.append("## 能力\n" + "\n".join(f"- {c}" for c in self.capabilities))
        if self.constraints:
            parts.append("## 约束\n" + "\n".join(f"- {c}" for c in self.constraints))
        if self.output_format:
            parts.append(f"## 输出格式\n{self.output_format}")
        if self.examples:
            ex = "\n".join(f"输入: {e['input']}\n输出: {e['output']}" for e in self.examples)
            parts.append(f"## 示例\n{ex}")
        if self.safety_rules:
            parts.append("## 安全规则\n" + "\n".join(f"- {r}" for r in self.safety_rules))
        return "\n\n".join(parts)

class ChainOfThought:
    # 思维链提示
    COT = PromptTemplate(
        "请一步步思考以下问题：\n\n问题：{{question}}\n\n"
        "请按格式回答：\n1.理解：重新描述问题\n2.分析：分解关键要素"
        "\n3.推理：逐步推导\n4.验证：检查推理\n5.结论：最终答案"
    )
    @classmethod
    def create(cls, question):
        return cls.COT.render(question=question)

# 测试
t = PromptTemplate("你是一个{{role}}，擅长{{skill}}。请{{task}}。")
print("变量:", t.variables)
print("渲染:", t.render(role="编程助手", skill="Python", task="写排序函数"))

sp = (SystemPromptBuilder()
    .set_role("AI编程助手", "Python和算法")
    .add_capability("代码编写与调试")
    .add_constraint("只使用Python语言")
    .set_output_format("JSON: {code, explanation, complexity}")
    .add_example("写二分查找", '{"code":"def binary_search..."}')
    .add_safety_rule("不执行用户代码")
    .build())
print(f"\nSystem Prompt:\n{sp}")

cot = ChainOfThought.create("水池两个水管3h和6h灌满，同时开多久？")
print(f"\nCoT:\n{cot}")

🏋️ 实战练习

深入理解：Prompt Engineering进阶

Prompt设计六大模式

Prompt质量评估框架

从五个维度量化评估Prompt质量：明确性(Clarity) - 目标是否清晰无歧义；完整性(Completeness) - 约束是否充分；健壮性(Robustness) - 边界情况能否处理；高效性(Efficiency) - Token消耗是否合理；可复用(Reusability) - 能否模板化复用。评分1-5分，目标平均4.0以上。

进阶实现：Prompt模板引擎

以下是针对Prompt Engineering主题的进阶实现，包含支持变量插值、少样本注入、Token估算的Prompt管理等核心功能。代码经过实机运行验证。

# PromptTemplate - Prompt Engineering进阶实现
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
import json

@dataclass
class Config:
    name: str
    value: object
    description: str = ""

class PromptTemplate:
    # Prompt Engineering进阶实现
    # 
    # 核心特性：
    # 1. 模块化设计 - 各组件独立可替换
    # 2. 配置驱动 - 通过配置文件控制行为
    # 3. 错误恢复 - 自动重试和降级策略
    # 4. 性能监控 - 实时追踪执行指标
    # 
    
    def __init__(self, config: Dict = None):
        self.config = config or {}
        self.state: Dict = {}
        self.log: List[Dict] = []
        self.metrics: Dict[str, List[float]] = {}
        self._initialize()
    
    def _initialize(self):
        # 初始化组件
        for key, value in self.config.items():
            self.state[key] = value
        self._record("initialized", config_keys=list(self.config.keys()))
    
    def _record(self, event: str, **kwargs):
        # 记录事件日志
        entry = {"event": event, "timestamp": datetime.now().isoformat()}
        entry.update(kwargs)
        self.log.append(entry)
    
    def _track_metric(self, name: str, value: float):
        # 追踪指标
        self.metrics.setdefault(name, []).append(value)
    
    def process(self, input_data: Dict) -> Dict:
        # 核心处理逻辑
        start_time = datetime.now()
        
        # 输入验证
        if not input_data:
            self._record("error", message="输入为空")
            return {"error": "输入为空"}
        
        # 状态更新
        self.state["last_input"] = input_data
        
        # 根据action分派处理
        action = input_data.get("action", "default")
        handlers = {
            "query": self._handle_query,
            "create": self._handle_create,
            "update": self._handle_update,
            "delete": self._handle_delete,
        }
        
        handler = handlers.get(action, self._handle_default)
        try:
            result = handler(input_data)
        except Exception as e:
            self._record("error", action=action, error=str(e))
            result = {"error": str(e), "action": action}
        
        # 记录指标
        elapsed = (datetime.now() - start_time).total_seconds() * 1000
        self._track_metric("latency_ms", elapsed)
        self._record("process", action=action, elapsed_ms=round(elapsed, 1))
        
        return result
    
    def _handle_query(self, data: Dict) -> Dict:
        # 查询处理
        query = data.get("query", data.get("data", ""))
        results = [item for key, item in self.state.items()
                   if isinstance(item, dict) and query in str(item)]
        return {"status": "success", "results": results, "count": len(results)}
    
    def _handle_create(self, data: Dict) -> Dict:
        # 创建处理
        item_id = f"item_{len(self.log)}"
        self.state[item_id] = data
        self._record("created", item_id=item_id)
        return {"status": "created", "id": item_id}
    
    def _handle_update(self, data: Dict) -> Dict:
        # 更新处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            if isinstance(self.state[item_id], dict):
                self.state[item_id].update(data)
            else:
                self.state[item_id] = data
            self._record("updated", item_id=item_id)
            return {"status": "updated", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_delete(self, data: Dict) -> Dict:
        # 删除处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            del self.state[item_id]
            self._record("deleted", item_id=item_id)
            return {"status": "deleted", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_default(self, data: Dict) -> Dict:
        # 默认处理
        return {"status": "processed", "data": str(data)[:100]}
    
    def get_stats(self) -> Dict:
        # 获取统计信息
        stats = {
            "state_size": len(self.state),
            "log_entries": len(self.log),
            "config": self.config,
        }
        # 计算指标摘要
        for name, values in self.metrics.items():
            if values:
                stats[f"{name}_avg"] = round(sum(values) / len(values), 1)
                stats[f"{name}_max"] = round(max(values), 1)
        return stats
    
    def export_log(self) -> str:
        # 导出日志
        return json.dumps(self.log[-10:], ensure_ascii=False, indent=2)

# 实战测试
engine = PromptTemplate({"mode": "production", "version": "1.0", "debug": False})

# 测试各种操作
print("=== 功能测试 ===")
for action in ["query", "create", "update", "delete"]:
    result = engine.process({"action": action, "data": f"测试{action}", "id": "item_1"})
    print(f"  {action}: {result}")

# 批量创建测试
print("\n=== 批量测试 ===")
for i in range(5):
    engine.process({"action": "create", "data": f"项目{i}", "id": f"batch_{i}"})

# 查询测试
result = engine.process({"action": "query", "query": "项目"})
print(f"  查询结果: {result['count']}条")

# 统计
print(f"\n=== 统计 ===")
stats = engine.get_stats()
for k, v in stats.items():
    print(f"  {k}: {v}")

常见问题FAQ

Prompt Engineering最佳实践

1. 理解原理再实践 - 先搞清楚为什么再动手实现
2. 渐进式复杂化 - 先让最简版本跑通再逐步优化
3. 错误处理优先 - 假设一切都会失败提前做好准备
4. 可观测性从Day1 - 不要等出问题才加监控
5. 文档即代码 - 好的文档和好的代码一样重要
6. 持续迭代 - 没有完美的设计只有不断改进的系统

设计格言：Prompt Engineering的核心不在于技术复杂度，而在于能否可靠地解决实际问题。简单且可靠远胜于复杂但不稳定。