Prompt Engineering 技术栈：从基础到 CoT/ToT 的完整体系

┌─────────────────────────────────────┐
│         Agent 编排与工具链            │  ← 系统架构层
├─────────────────────────────────────┤
│      RAG / Fine-tuning / 微调       │  ← 能力增强层
├─────────────────────────────────────┤
│      Prompt Engineering 技术栈       │  ← 交互基座层  ★
├─────────────────────────────────────┤
│           LLM 基础模型能力           │  ← 底座推理层
└─────────────────────────────────────┘

# 弱角色设定 — 模型行为不可预测
system_prompt = "你是一个助手。"

# 强角色设定 — 模型行为被精确锚定
system_prompt = """
你是一位拥有 15 年经验的高级后端架构师，专注于分布式系统设计。
你倾向于用简洁的技术语言回答问题，在给出建议前会先分析现有架构的瓶颈。
你不会使用比喻或类比，直接给出技术方案。
"""

# 模糊任务 — 输出不可控
task = "分析一下这个系统"

# 精确任务 — 输出可预期
task = """
分析以下电商订单系统的数据库 Schema，识别以下问题：
1. 是否存在 N+1 查询的表结构设计缺陷
2. 索引覆盖度是否满足高并发读写需求
3. 是否有数据一致性风险（如分库分表场景下的跨表事务）
"""

constraints = """
约束条件：
- 不要猜测不确定的技术细节，无法确认时明确标注"需要进一步确认"
- 不要给出超过 3 个解决方案，只推荐你最有信心的方案
- 不要忽略安全方面的考量
- 所有性能数据必须附带假设前提
"""

format_spec = """
输出格式要求（严格 JSON）：
{
  "analysis": "问题分析",
  "root_cause": "根因判断",
  "severity": "critical|high|medium|low",
  "recommendations": [
    {
      "action": "具体建议",
      "priority": 1,
      "estimated_effort": "人天估算"
    }
  ]
}
"""

from openai import OpenAI
client = OpenAI()

# ============================================
# Zero-shot：适合简单的结构化输出
# ============================================
zero_shot_response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "你是一个日志分析专家。"},
        {"role": "user", "content": "将以下非结构化日志转换为 JSON 格式：\n"
         "2024-01-15 10:23:45 ERROR [auth-service] Failed to validate token: expired"}
    ]
)

# ============================================
# Few-shot：复杂格式时，示例比描述更高效
# ============================================
few_shot_response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "你是一个日志分析专家，将非结构化日志解析为结构化 JSON。"},
        {"role": "user", "content": "2024-01-15 10:23:45 INFO [api-gateway] Request processed: GET /users/123 - 200 OK (45ms)"},
        {"role": "assistant", "content": """```json
{
  "timestamp": "2024-01-15T10:23:45",
  "level": "INFO",
  "service": "api-gateway",
  "message": "Request processed",
  "metadata": {
    "method": "GET",
    "path": "/users/123",
    "status_code": 200,
    "latency_ms": 45
  }
}
```"""},
        {"role": "user", "content": "2024-01-15 10:23:46 WARN [payment-service] Stripe webhook signature verification failed for event evt_123456"},
        {"role": "assistant", "content": """```json
{
  "timestamp": "2024-01-15T10:23:46",
  "level": "WARN",
  "service": "payment-service",
  "message": "Stripe webhook signature verification failed",
  "metadata": {
    "event_id": "evt_123456",
    "provider": "stripe"
  }
}
```"""},
        {"role": "user", "content": "2024-01-15 10:23:47 ERROR [database] Connection pool exhausted: active=50 max=50 waiting=12 timeout=30s"}
    ]
)

任务复杂度评估
│
├── 是否需要特定输出格式？
│   ├── 否 → Zero-shot
│   └── 是 ↓
│
├── 格式是否可以仅用文字描述清楚？
│   ├── 是 → Zero-shot + Format constraints
│   └── 否 ↓
│
├── 示例数量需求？
│   ├── 1 个示例即可锚定 → One-shot
│   └── 需要多维度覆盖 → Few-shot (3-5 个)
│
└── 输出容错要求？
    ├── 宽容（允许格式微调） → 减少示例数
    └── 严格（必须精确匹配） → 增加示例数 + Constraints

standard_cot_response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "你是一个数学推理专家。"},
        {"role": "user", "content": """
一个农场有 15 只鸡和 8 只鸭。鸡每天下蛋，每只鸡每天下 1 个蛋。
鸭每 3 天下 1 个蛋。农场主人每周收集所有蛋，每周能收到多少个蛋？

让我们一步一步思考：

第一步：计算鸡每周的产蛋量。
- 鸡有 15 只，每只每天下 1 个蛋
- 鸡每天产蛋：15 × 1 = 15 个
- 鸡每周产蛋：15 × 7 = 105 个

第二步：计算鸭每周的产蛋量。
- 鸭有 8 只，每 3 天下 1 个蛋
- 鸭每只每周产蛋：7 ÷ 3 ≈ 2.33 个（向下取整为 2 个，因为需要完整 3 天周期）
- 鸭每周产蛋：8 × 2 = 16 个

第三步：计算总量。
- 总产蛋量：105 + 16 = 121 个

答案：农场每周能收到 121 个蛋。
"""},
        {"role": "user", "content": """
一个商店有 20 件衬衫，每件成本 15 元，售价 30 元。
每天卖出 3 件，每件衬衫运营成本 2 元/天。
问 10 天后，商店的总利润是多少？

让我们一步一步思考：
"""
        }
    ]
)

# Zero-shot CoT 的核心：在任何问题后追加 "Let's think step by step"
zero_shot_cot_response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "user", "content": """
小明有 5 个苹果，给了小红 2 个，又买了 3 个。
然后他把苹果平均分给了 3 个朋友。
问每个朋友得到几个苹果？剩下几个？

Let's think step by step.
"""}
    ]
)

# Auto-CoT 的实现思路
def auto_cot(question: str, clusters: list[list[str]], model: str = "gpt-4o") -> str:
    """
    Auto-CoT 流程：
    1. 将问题池聚类（使用 Sentence-BERT 等嵌入模型）
    2. 从每个聚类中选择代表性问题
    3. 用 Zero-shot CoT 为每个代表性问题生成推理链
    4. 将生成的推理链作为 Few-shot 示例
    """
    demos = []
    for cluster in clusters:
        representative_question = cluster[0]
        reasoning_chain = client.chat.completions.create(
            model=model,
            messages=[
                {"role": "user", "content": f"""
{representative_question}

Let's think step by step.
"""}
            ]
        ).choices[0].message.content
        demos.append((representative_question, reasoning_chain))

    prompt = "以下是一些示例：\n\n"
    for q, r in demos:
        prompt += f"问题：{q}\n{r}\n\n"
    prompt += f"问题：{question}\nLet's think step by step.\n"

    final_response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}]
    )
    return final_response.choices[0].message.content

complex_cot_response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": """
你是一个严谨的推理专家。在每一步推理后，你必须：
1. 给出推理结论
2. 验证该结论的合理性（合理性检查）
3. 如果验证不通过，回退修正
请用以下格式输出：

## 步骤 N：<步骤描述>
**推理**：...
**验证**：这个结论是否合理？[合理/不合理] 理由：...
**修正**（如需要）：...
"""},
        {"role": "user", "content": """
一个水池有两根进水管 A 和 B，以及一根排水管 C。
- A 管单独注满需要 6 小时
- B 管单独注满需要 8 小时  
- C 管单独排空需要 12 小时
- 三管同时打开，多少小时能注满水池？

请用分步验证的格式回答。
"""}
    ]
)

                    问题
                   / | \
                 /   |   \
              思路A  思路B  思路C
              / \     |    / \
           A1  A2    B1   C1  C2
           |        / \        |
          A1验证  B1a B1b    C2验证
           ✗      ✓    ✗     ✓
                    ↓           ↓
                  继续展开     继续展开

from dataclasses import dataclass

@dataclass
class ThoughtNode:
    content: str
    score: float = 0.0
    children: list = None

def tot_solve(problem: str, model: str = "gpt-4o", max_depth: int = 3, branch_width: int = 3):
    """
    ToT 求解器实现
    """
    root = ThoughtNode(content=problem)

    def generate_thoughts(node: ThoughtNode) -> list[ThoughtNode]:
        prompt = f"""
当前问题：{problem}
已有的推理路径：{node.content}
请生成 {branch_width} 个不同的下一步推理方向，每个方向简要说明思路。
"""
        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}]
        )
        thoughts = response.choices[0].message.content.split("\n\n")
        return [ThoughtNode(content=t, children=[]) for t in thoughts if t.strip()]

    def evaluate_thought(node: ThoughtNode) -> float:
        prompt = f"""
评估以下推理步骤的质量（0-10 分）：
问题：{problem}
推理路径：{node.content}

评分标准：逻辑严密性、方向正确性、信息增量。
只输出一个数字分数。
"""
        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}]
        )
        try:
            return float(response.choices[0].message.content.strip())
        except ValueError:
            return 0.0

    frontier = [root]
    for depth in range(max_depth):
        candidates = []
        for node in frontier:
            children = generate_thoughts(node)
            for child in children:
                child.score = evaluate_thought(child)
            candidates.extend(children)

        candidates.sort(key=lambda x: x.score, reverse=True)
        frontier = candidates[:branch_width]

    best = max(frontier, key=lambda x: x.score)
    return best.content

from collections import Counter
import re

def self_consistency_solve(
    question: str,
    model: str = "gpt-4o",
    num_samples: int = 5,
    temperature: float = 0.7
) -> dict:
    answers = []
    reasoning_paths = []

    for _ in range(num_samples):
        response = client.chat.completions.create(
            model=model,
            temperature=temperature,
            messages=[
                {"role": "user", "content": f"""
{question}

Let's think step by step.
请在最后一行用 "答案：XXX" 的格式给出最终答案。
"""}
            ]
        )
        content = response.choices[0].message.content
        reasoning_paths.append(content)

        answer_match = re.search(r"答案[：:]\s*(.+)", content)
        if answer_match:
            answers.append(answer_match.group(1).strip())

    if not answers:
        return {"answer": "无法确定", "confidence": 0.0, "paths": reasoning_paths}

    vote_counts = Counter(answers)
    best_answer, count = vote_counts.most_common(1)[0]

    return {
        "answer": best_answer,
        "confidence": count / len(answers),
        "vote_distribution": dict(vote_counts),
        "paths": reasoning_paths
    }

result = self_consistency_solve("一个班有 30 个学生，其中 18 人喜欢数学，15 人喜欢物理，8 人两者都喜欢。多少人两者都不喜欢？")

Thought → Action → Observation → Thought → Action → Observation → ... → Final Answer

import json

def react_agent(question: str, tools: dict, model: str = "gpt-4o", max_steps: int = 10):
    tool_descriptions = "\n".join([
        f"- {name}: {desc}" for name, desc in tools.items()
    ])

    system_prompt = f"""
你是一个 ReAct 智能体。你通过交替推理和行动来回答问题。

可用工具：
{tool_descriptions}

严格按以下格式输出（每次只输出一个步骤）：

Thought: <你的推理>
Action: <工具名>
Action Input: <工具参数>

收到 Observation 后继续推理。
当可以给出最终答案时，输出：
Thought: <最终推理>
Final Answer: <答案>
"""

    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": question}
    ]

    for step in range(max_steps):
        response = client.chat.completions.create(
            model=model, messages=messages
        )
        assistant_msg = response.choices[0].message.content
        messages.append({"role": "assistant", "content": assistant_msg})

        if "Final Answer:" in assistant_msg:
            final_answer = assistant_msg.split("Final Answer:")[-1].strip()
            return {"answer": final_answer, "steps": step + 1}

        action_match = re.search(r"Action:\s*(.+)", assistant_msg)
        input_match = re.search(r"Action Input:\s*(.+)", assistant_msg)

        if action_match and input_match:
            tool_name = action_match.group(1).strip()
            tool_input = input_match.group(1).strip()

            if tool_name in tools:
                observation = tools[tool_name](tool_input)
            else:
                observation = f"Error: Tool '{tool_name}' not found"

            messages.append({
                "role": "user",
                "content": f"Observation: {observation}"
            })

    return {"answer": "达到最大步数限制，未能得出最终答案", "steps": max_steps}

tools = {
    "search": lambda q: f"搜索结果: 关于'{q}'的最新信息...",
    "calculate": lambda expr: f"计算结果: {eval(expr)}",
    "lookup": lambda term: f"查询结果: {term} 的定义是..."
}

result = react_agent("2024 年诺贝尔物理学奖得主是谁？他们获得博士学位的大学排名如何？", tools)

structured_prompt = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": """
你必须严格按以下 JSON Schema 输出，不要包含任何其他文字：

{
  "severity": "critical" | "high" | "medium" | "low",
  "category": "string",
  "summary": "string (50字以内)",
  "affected_systems": ["string"],
  "remediation_steps": ["string"]
}
"""},
        {"role": "user", "content": "分析以下安全事件：服务器 10.0.1.5 出现异常外联行为，疑似 C2 通信，涉及进程 svchost.exe 的异常子进程。"}
    ]
)

from pydantic import BaseModel

class SecurityAlert(BaseModel):
    severity: str
    category: str
    summary: str
    affected_systems: list[str]
    remediation_steps: list[str]

response = client.beta.chat.completions.parse(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "分析安全事件并输出结构化报告。"},
        {"role": "user", "content": "服务器 10.0.1.5 出现异常外联行为，疑似 C2 通信。"}
    ],
    response_format=SecurityAlert
)

alert = response.choices[0].message.parsed

table_response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "user", "content": """
比较 Redis、Memcached 和 MemSQL 三种缓存方案。
用 Markdown 表格输出，包含以下列：方案名称、数据结构、持久化、集群支持、适用场景。
"""}
    ]
)

# prompt_registry.py — Prompt 版本化管理

from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class PromptVersion:
    version: str
    system_prompt: str
    user_template: str
    model: str
    temperature: float
    metadata: dict = field(default_factory=dict)
    created_at: str = field(default_factory=lambda: datetime.now().isoformat())

PROMPT_REGISTRY = {
    "log-parser": {
        "v1.0": PromptVersion(
            version="v1.0",
            system_prompt="你是一个日志分析助手。",
            user_template="分析以下日志：\n{log_content}",
            model="gpt-4o",
            temperature=0.3,
            metadata={"author": "team", "baseline_accuracy": 0.72}
        ),
        "v1.1": PromptVersion(
            version="v1.1",
            system_prompt="你是一个资深日志分析专家。将非结构化日志解析为标准 JSON 格式。",
            user_template="分析以下日志并输出 JSON：\n{log_content}\n\n输出格式：{format_spec}",
            model="gpt-4o",
            temperature=0.2,
            metadata={"author": "team", "baseline_accuracy": 0.89,
                      "changes": "增加了角色设定和格式约束"}
        ),
    }
}

def get_prompt(task_name: str, version: str = "latest") -> PromptVersion:
    versions = PROMPT_REGISTRY[task_name]
    if version == "latest":
        return list(versions.values())[-1]
    return versions[version]

import random
import hashlib
from collections import defaultdict

class PromptABTest:
    def __init__(self, task_name: str):
        self.task_name = task_name
        self.results = defaultdict(lambda: {"correct": 0, "total": 0, "latencies": []})

    def select_version(self, user_id: str, variants: list[str]) -> str:
        hash_val = int(hashlib.md5(f"{self.task_name}:{user_id}".encode()).hexdigest(), 16)
        return variants[hash_val % len(variants)]

    def record_result(self, version: str, correct: bool, latency_ms: float):
        self.results[version]["total"] += 1
        if correct:
            self.results[version]["correct"] += 1
        self.results[version]["latencies"].append(latency_ms)

    def get_report(self) -> dict:
        report = {}
        for version, data in self.results.items():
            report[version] = {
                "accuracy": data["correct"] / data["total"] if data["total"] > 0 else 0,
                "total_samples": data["total"],
                "avg_latency_ms": sum(data["latencies"]) / len(data["latencies"]) if data["latencies"] else 0
            }
        return report

# 评估指标体系
EVALUATION_METRICS = {
    "accuracy": "输出结果的正确率",
    "format_compliance": "格式符合要求的比例",
    "completeness": "输出是否覆盖了所有要求的信息点",
    "relevance": "输出与问题的相关性评分",
    "token_efficiency": "有效输出 token / 总 token 消耗",
    "latency_p50_p99": "响应延迟分布",
    "hallucination_rate": "幻觉信息占比（需人工标注或自动化检测）"
}

1. 建立基准（Baseline）
   ├── 收集 50-100 个代表性测试用例
   ├── 记录当前 Prompt 版本的各项指标
   └── 确定优化目标（精度？延迟？成本？）

2. 分析错误模式
   ├── 分类错误类型（格式错误、逻辑错误、幻觉等）
   ├── 识别高频错误模式
   └── 确定最优先修复的错误类型

3. 迭代优化
   ├── 针对性调整 Prompt 要素（增加示例、细化约束等）
   ├── A/B 测试新旧版本
   ├── 量化对比各项指标
   └── 如果新版本显著优于旧版本，升级版本号

4. 监控与回归
   ├── 生产环境持续监控各项指标
   ├── 模型升级时重新验证 Prompt 效果
   └── 定期清理过时的 Prompt 版本

Prompt Engineering 技术栈
│
├── 基础技术
│   ├── 角色设定（Role Prompting）
│   ├── 指令工程（Instruction Engineering）
│   ├── 输出格式控制（Output Formatting）
│   └── 分隔符与结构化（Delimiters & Structure）
│
├── 示例驱动
│   ├── Zero-shot（零样本）
│   ├── One-shot（单样本）
│   ├── Few-shot（少样本）
│   └── Dynamic Few-shot（动态选择示例）
│
├── 推理增强
│   ├── Chain-of-Thought（CoT）
│   │   ├── Standard CoT（标准思维链）
│   │   ├── Zero-shot CoT（零样本思维链）
│   │   ├── Auto-CoT（自动思维链）
│   │   └── Complex CoT（复杂思维链）
│   │
│   ├── Tree-of-Thought（ToT）
│   │   ├── BFS 策略（广度优先）
│   │   └── DFS 策略（深度优先）
│   │
│   ├── Self-Consistency（自一致性）
│   ├── Graph-of-Thought（GoT）
│   └── Least-to-Most（从简到繁）
│
├── 推理与行动
│   ├── ReAct（推理+行动交替）
│   ├── Toolformer（工具学习）
│   └── Reflexion（自我反思）
│
├── 鲁棒性增强
│   ├── Self-Consistency（多路径投票）
│   ├── Self-Refine（自我修正）
│   ├── Generated Knowledge（生成知识增强）
│   └── Directional Stimulus（方向性刺激）
│
└── 工程化实践
    ├── Prompt 版本管理
    ├── A/B 测试框架
    ├── 自动化评估管线
    └── Prompt 模板库