Prompt Engineering 进阶实战：结构化输出、自一致性采样与思维链编排技术详解

import json
from pydantic import BaseModel, Field
from openai import OpenAI

client = OpenAI(api_key="sk-your-key")

# 第一步：用 Pydantic 定义输出结构
class BugReport(BaseModel):
    title: str = Field(description="Bug 标题，一句话描述")
    severity: str = Field(description="严重程度：critical/major/minor")
    component: str = Field(description="受影响模块")
    steps: list[str] = Field(description="复现步骤")
    suggested_fix: str = Field(description="建议修复方案")

# 第二步：调用支持结构化输出的 API
response = client.beta.chat.completions.parse(
    model="gpt-4o",
    messages=[
        {
            "role": "system",
            "content": "你是资深 QA 工程师，将用户描述的 Bug 转换为结构化报告。"
        },
        {
            "role": "user",
            "content": "登录页面点击'记住我'后，刷新页面会自动登出。Chrome 130，生产环境。"
        }
    ],
    response_format=BugReport,  # 核心参数：直接传 Pydantic 模型
)

# 第三步：直接拿到类型安全的对象
report: BugReport = response.choices[0].message.parsed
print(f"[{report.severity}] {report.title}")
print(f"模块: {report.component}")
print(f"复现步骤: {' -> '.join(report.steps)}")
print(f"建议修复: {report.suggested_fix}")

import collections
from openai import OpenAI

client = OpenAI(api_key="sk-your-key")

COT_PROMPT = """请逐步推理并解答以下问题。
问题：{question}
要求：
1. 先写出推理过程（reasoning）
2. 最后一行只写最终答案，格式为 ANSWER: <数字>
"""

def single_reasoning(question: str, temperature: float = 0.7) -> str:
    """单次思维链推理"""
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": COT_PROMPT.format(question=question)}],
        temperature=temperature,  # 关键：用较高温度增加推理路径多样性
        max_tokens=1024,
    )
    return resp.choices[0].message.content

def extract_answer(response: str) -> str | None:
    """从推理结果中提取最终答案"""
    for line in response.strip().split("\n"):
        if line.strip().upper().startswith("ANSWER:"):
            return line.split(":", 1)[1].strip()
    return None

def self_consistency_solve(question: str, n: int = 5) -> str:
    """自一致性采样：多路推理 + 多数投票"""
    results = []
    for _ in range(n):
        output = single_reasoning(question)
        answer = extract_answer(output)
        if answer:
            results.append(answer)

    if not results:
        return "无法确定"

    # 多数投票
    counter = collections.Counter(results)
    best_answer, votes = counter.most_common(1)[0]
    confidence = votes / len(results)
    print(f"投票结果: {dict(counter)} | 置信度: {confidence:.0%}")
    return best_answer

# 测试
question = "一个水池有进水管和出水管。进水管每小时注水 3 吨，出水管每小时排水 1.5 吨。水池容量 20 吨，当前有 5 吨水。问几小时后水池满？"
answer = self_consistency_solve(question, n=5)
print(f"最终答案: {answer} 小时")

import json
from openai import OpenAI

client = OpenAI(api_key="sk-your-key")

def llm(system: str, user: str) -> str:
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system},
            {"role": "user", "content": user},
        ],
        temperature=0.3,
    )
    return resp.choices[0].message.content

def orchestrated_analysis(user_query: str) -> dict:
    """三步编排：分解 → 推理 → 总结"""

    # 步骤1：任务分解
    step1 = llm(
        "你是任务规划器。将用户问题拆解为 2-4 个子问题，输出 JSON 数组。",
        f"用户问题：{user_query}\n输出格式：[\"子问题1\", \"子问题2\", ...]"
    )
    sub_questions = json.loads(step1)
    print(f"[规划] 子问题: {sub_questions}")

    # 步骤2：逐个推理
    sub_answers = []
    for i, sq in enumerate(sub_questions):
        answer = llm(
            "你是领域专家。针对子问题给出简洁、准确的回答。",
            f"子问题：{sq}\n背景：原始问题是「{user_query}」"
        )
        sub_answers.append({"question": sq, "answer": answer})
        print(f"[推理 {i+1}] 完成")

    # 步骤3：综合总结
    final = llm(
        "你是综合分析师。根据各子问题的回答，给出对原始问题的完整回答。",
        f"原始问题：{user_query}\n子问题与回答：{json.dumps(sub_answers, ensure_ascii=False)}"
    )

    return {
        "sub_questions": sub_questions,
        "sub_answers": sub_answers,
        "final_answer": final,
    }

# 测试
result = orchestrated_analysis("我们公司该不该从 MySQL 迁移到 PostgreSQL？")
print(f"\n最终建议:\n{result['final_answer']}")