高级 Prompt 工程技术：Chain-of-Thought、Few-Shot 与结构化提示词的深度实践

import openai

def zero_shot_cot(question: str, model: str = "gpt-4") -> str:
    """
    Zero-Shot Chain-of-Th prompting.
    只需在问题后加 'Let's think step by step' 即可。
    """
    prompt = f"""请回答以下问题。在给出最终答案之前，请先一步步分析。

问题：{question}

请逐步思考并给出答案："""

    response = openai.ChatCompletion.create(
        model=model,
        messages=[
            {"role": "system", "content": "你是一个善于逻辑推理的 AI 助手。"},
            {"role": "user", "content": prompt}
        ],
        temperature=0.0,  # 推理任务用低温度
        max_tokens=1024
    )
    return response.choices[0].message.content


# 测试数学题
question = """
小明有 120 元，买了一本书花了 35 元，
又买了 3 支笔，每支 8 元。
请问他还剩多少钱？
"""
result = zero_shot_cot(question)
print("=== 推理过程 ===")
print(result)

import openai

# 精心设计的 Few-Shot CoT 示例
COT_EXAMPLES = [
    {
        "question": "餐厅有 23 个苹果，用 20 个做了午餐，又买了 6 个。现在有多少个？",
        "reasoning": "1. 初始苹果数量：23 个\n2. 用掉 20 个后剩余：23 - 20 = 3 个\n3. 又买了 6 个：3 + 6 = 9 个\n4. 最终苹果数量：9 个",
        "answer": "9"
    },
    {
        "question": "一个剧团有 4 名舞者，每场演出需要 3 人，每场演出持续 2 小时。如果要让每个舞者表演相同时长，每场演出之间休息 30 分钟，完成一轮全员等时演出需要多长时间？",
        "reasoning": "1. 每场演出 3 人，每人演 2 小时，每场总表演时长：3 × 2 = 6 小时\n2. 4 名舞者每人演相同时长，需要 4 × 2 = 8 小时总表演\n3. 需要场次：8 / 6 = 1.33，向上取整为 2 场\n4. 2 场演出时间：2 × 2 = 4 小时\n5. 1 次休息时间：30 分钟 = 0.5 小时\n6. 总时长：4 + 0.5 = 4.5 小时",
        "answer": "4.5 小时"
    },
    {
        "question": "一辆公交车上有 47 人，到站后下去了 15 人，又上来了 8 人。下一站下去了 12 人，上来了 5 人。现在车上有多少人？",
        "reasoning": "1. 初始人数：47 人\n2. 第一站后：47 - 15 + 8 = 40 人\n3. 第二站后：40 - 12 + 5 = 33 人\n4. 现在车上人数：33 人",
        "answer": "33"
    }
]

def build_few_shot_cot_prompt(question: str) -> str:
    """构建 Few-Shot CoT Prompt"""
    parts = []
    
    # 添加示例
    for ex in COT_EXAMPLES:
        parts.append(f"问题：{ex['question']}")
        parts.append(f"推理过程：{ex['reasoning']}")
        parts.append(f"答案：{ex['answer']}")
        parts.append("---")
    
    # 添加目标问题
    parts.append(f"问题：{question}")
    parts.append("推理过程：")
    
    return "\n".join(parts)


def few_shot_cot(question: str, model: str = "gpt-4") -> dict:
    """执行 Few-Shot CoT 推理"""
    prompt = build_few_shot_cot_prompt(question)
    
    response = openai.ChatCompletion.create(
        model=model,
        messages=[
            {"role": "system", "content": "你是一个数学推理助手。请按照示例的格式，先写出详细的推理过程，再给出答案。"},
            {"role": "user", "content": prompt}
        ],
        temperature=0.0,
        max_tokens=1024
    )
    
    result = response.choices[0].message.content
    
    # 解析推理过程和答案
    if "答案：" in result:
        reasoning, answer = result.rsplit("答案：", 1)
        return {
            "reasoning": reasoning.strip(),
            "answer": answer.strip()
        }
    return {"reasoning": result, "answer": "未能提取答案"}


# 测试
test_question = "商店有 56 瓶饮料，上午卖了 18 瓶，下午进了 12 瓶，晚上又卖了 9 瓶。还剩多少瓶？"
result = few_shot_cot(test_question)
print(f"推理：{result['reasoning']}")
print(f"答案：{result['answer']}")

import openai
from collections import Counter
from concurrent.futures import ThreadPoolExecutor, as_completed
import re


def extract_answer(text: str) -> str:
    """从推理文本中提取最终答案"""
    # 尝试匹配 "答案是 X" 或 "answer: X" 或最后的数字
    patterns = [
        r"答案是[：:]\s*(.+?)(?:。|$)",
        r"answer[：:]\s*(.+?)(?:。|$)",
        r"所以答案是[：:]\s*(.+?)(?:。|$)",
        r"最终答案[：:]\s*(.+?)(?:。|$)",
    ]
    for pattern in patterns:
        match = re.search(pattern, text)
        if match:
            return match.group(1).strip()
    
    # 回退：提取最后的数字
    numbers = re.findall(r'\d+\.?\d*', text)
    return numbers[-1] if numbers else "无法提取"


def single_cot_run(question: str, model: str, seed: int) -> str:
    """执行单次 CoT 推理"""
    response = openai.ChatCompletion.create(
        model=model,
        messages=[
            {"role": "system", "content": "请一步步推理这个问题，最后给出答案。"},
            {"role": "user", "content": f"问题：{question}\n请逐步思考："}
        ],
        temperature=0.7,  # 需要一定的随机性
        max_tokens=512,
        seed=seed
    )
    return response.choices[0].message.content


def self_consistency_cot(
    question: str,
    model: str = "gpt-4",
    n_runs: int = 5,
    max_workers: int = 5
) -> dict:
    """
    Self-Consistency Chain-of-Thought
    
    参数：
        question: 需要推理的问题
        model: 使用的模型
        n_runs: 推理次数
        max_workers: 并行线程数
    
    返回：
        包含投票结果、各答案出现次数、详细推理的字典
    """
    results = []
    
    # 并行执行多次推理
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(single_cot_run, question, model, seed=i): i
            for i in range(n_runs)
        }
        
        for future in as_completed(futures):
            seed = futures[future]
            try:
                reasoning = future.result()
                answer = extract_answer(reasoning)
                results.append({
                    "seed": seed,
                    "reasoning": reasoning,
                    "answer": answer
                })
            except Exception as e:
                print(f"Run {seed} failed: {e}")
    
    # 投票
    answers = [r["answer"] for r in results]
    vote_count = Counter(answers)
    best_answer, vote_tally = vote_count.most_common(1)[0]
    
    # 找到最佳推理
    best_runs = [r for r in results if r["answer"] == best_answer]
    
    return {
        "question": question,
        "best_answer": best_answer,
        "vote_tally": f"{vote_tally}/{n_runs}",
        "confidence": vote_tally / n_runs,
        "all_answers": dict(vote_count),
        "best_reasoning": best_runs[0]["reasoning"],
        "all_results": results
    }


# 使用示例
question = "一个农场有鸡和兔子，共有 35 个头，94 只脚。问鸡和兔子各有多少只？"
result = self_consistency_cot(question, n_runs=5)

print(f"问题：{result['question']}")
print(f"最佳答案：{result['best_answer']}")
print(f"投票结果：{result['vote_tally']}")
print(f"置信度：{result['confidence']:.0%}")
print(f"所有答案分布：{result['all_answers']}")
print(f"\n最佳推理过程：")
print(result['best_reasoning'])

import openai
import re


# ReAct Prompt 模板
REACT_TEMPLATE = """你是一个智能助手，通过思考和行动来回答问题。

你可以使用以下工具：
- search[query]: 搜索网络获取信息
- calculate[expression]: 计算数学表达式
- knowledge[topic]: 获取某个主题的常识

请按照以下格式思考：
Thought: [你的思考，分析当前情况]
Action: [你决定使用的工具和参数]
Observation: [工具返回的结果]
...（重复 Thought/Action/Observation）
Thought: [最终推理]
Final Answer: [最终答案]

下面是一个示例：

Question: 苹果公司成立于哪年？截至 2024 年有多少员工？
Thought: 我需要搜索苹果公司的成立时间和员工数量。
Action: search[苹果公司 成立时间 员工数量 2024]
Observation: 苹果公司由 Steve Jobs 等人于 1976 年 4 月 1 日创立。截至 2024 年，苹果拥有约 164,000 名全职员工。
Thought: 我已经找到了答案。
Final Answer: 苹果公司成立于 1976 年，截至 2024 年有约 164,000 名员工。

现在请回答：

Question: {question}
"""


class ReActAgent:
    """简单的 ReAct Agent 实现"""
    
    def __init__(self, model: str = "gpt-4"):
        self.model = model
        self.max_iterations = 5
    
    def _call_llm(self, prompt: str) -> str:
        """调用 LLM"""
        response = openai.ChatCompletion.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0,
            max_tokens=512
        )
        return response.choices[0].message.content
    
    def _parse_action(self, text: str) -> tuple[str, str] | None:
        """解析 Action: tool[query]"""
        match = re.search(r'Action:\s*(\w+)\[(.+?)\]', text)
        if match:
            return match.group(1), match.group(2)
        return None
    
    def _execute_action(self, tool: str, query: str) -> str:
        """模拟工具执行（实际中应接入真实工具）"""
        tools = {
            "search": lambda q: f"[搜索结果] 关于'{q}'的信息...",
            "calculate": lambda q: str(eval(q)),
            "knowledge": lambda q: f"[知识] {q} 的常识性信息..."
        }
        return tools.get(tool, lambda q: "未知工具")(query)
    
    def run(self, question: str) -> str:
        """执行 ReAct 推理"""
        history = REACT_TEMPLATE.format(question=question)
        
        for i in range(self.max_iterations):
            # 获取 LLM 响应
            response = self._call_llm(history)
            history += "\n" + response
            
            # 解析 Action
            action = self._parse_action(response)
            
            if action is None:
                # 没有 Action，说明已经有 Final Answer
                break
            
            tool, query = action
            print(f"Iteration {i+1}:")
            print(f"  Thought: {response.split('Thought:')[1].split('Action:')[0] if 'Action:' in response else response}")
            print(f"  Action: {tool}[{query}]")
            
            # 执行工具
            observation = self._execute_action(tool, query)
            print(f"  Observation: {observation[:100]}...")
            
            # 添加 Observation 到历史
            history += f"\nObservation: {observation}\n"
        
        # 提取 Final Answer
        final_match = re.search(r'Final Answer:\s*(.+)', history, re.DOTALL)
        return final_match.group(1).strip() if final_match else "未能得出结论"


# 使用示例
agent = ReActAgent()
result = agent.run("Python 3.12 比 3.11 快多少百分比？")
print(f"\nFinal Answer: {result}")

from enum import Enum
from dataclasses import dataclass
from typing import Any
import json
import openai


class TaskType(Enum):
    MATH = "math"
    RESEARCH = "research"
    CREATIVE = "creative"
    EXTRACTION = "extraction"
    SIMPLE = "simple"


@dataclass
class PromptResult:
    answer: str
    confidence: float
    reasoning: str
    task_type: TaskType
    retry_count: int = 0


class TaskClassifier:
    """简单任务分类器（实际中可用 LLM 判断）"""
    
    MATH_KEYWORDS = ["计算", "多少", "百分比", "平均", "总和", "概率", "方程"]
    RESEARCH_KEYWORDS = ["搜索", "最新", "趋势", "比较", "评价"]
    CREATIVE_KEYWORDS = ["写", "创作", "故事", "诗歌", "文案"]
    EXTRACTION_KEYWORDS = ["提取", "解析", "JSON", "结构化", "表格"]
    
    @classmethod
    def classify(cls, question: str) -> TaskType:
        q = question.lower()
        if any(k in q for k in cls.MATH_KEYWORDS):
            return TaskType.MATH
        if any(k in q for k in cls.RESEARCH_KEYWORDS):
            return TaskType.RESEARCH
        if any(k in q for k in cls.CREATIVE_KEYWORDS):
            return TaskType.CREATIVE
        if any(k in q for k in cls.EXTRACTION_KEYWORDS):
            return TaskType.EXTRACTION
        return TaskType.SIMPLE


class PromptRouter:
    """生产级 Prompt 路由器"""
    
    def __init__(self, model: str = "gpt-4"):
        self.model = model
        self.max_retries = 2
    
    def run_cot_with_consistency(
        self, question: str, n: int = 3
    ) -> PromptResult:
        """数学/逻辑任务：CoT + Self-Consistency"""
        # 简化版：多次调用取多数
        answers = []
        for _ in range(n):
            resp = openai.ChatCompletion.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "请一步步推理这个问题。"},
                    {"role": "user", "content": f"问题：{question}\n请逐步思考："}
                ],
                temperature=0.7
            )
            answers.append(resp.choices[0].message.content)
        
        # 简单投票：找最长的答案（通常推理更详细）
        best = max(answers, key=len)
        return PromptResult(
            answer=best,
            confidence=0.85,
            reasoning="Self-Consistency CoT",
            task_type=TaskType.MATH
        )
    
    def run_structured_extraction(
        self, question: str, schema: dict
    ) -> PromptResult:
        """结构化数据提取"""
        prompt = f"""请从以下文本中提取信息，并严格按照 JSON Schema 格式输出。

Schema: {json.dumps(schema, ensure_ascii=False)}

文本：{question}

请只输出 JSON，不要其他内容。"""
        
        resp = openai.ChatCompletion.create(
            model=self.model,
            messages=[
                {"role": "system", "content": "你是一个数据提取助手。只输出有效的 JSON。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.0,
            response_format={"type": "json_object"}
        )
        
        return PromptResult(
            answer=resp.choices[0].message.content,
            confidence=0.95,
            reasoning="Structured JSON extraction",
            task_type=TaskType.EXTRACTION
        )
    
    def process(self, question: str, **kwargs) -> PromptResult:
        """主入口：自动分类并选择最佳策略"""
        task_type = TaskClassifier.classify(question)
        
        for attempt in range(self.max_retries):
            try:
                if task_type == TaskType.MATH:
                    return self.run_cot_with_consistency(question)
                elif task_type == TaskType.EXTRACTION:
                    schema = kwargs.get("schema", {"type": "object"})
                    return self.run_structured_extraction(question, schema)
                else:
                    # 默认：直接回答
                    resp = openai.ChatCompletion.create(
                        model=self.model,
                        messages=[
                            {"role": "system", "content": "你是一个知识渊博的助手。"},
                            {"role": "user", "content": question}
                        ],
                        temperature=0.7
                    )
                    return PromptResult(
                        answer=resp.choices[0].message.content,
                        confidence=0.7,
                        reasoning="Direct answer",
                        task_type=task_type,
                        retry_count=attempt
                    )
            except Exception as e:
                print(f"Attempt {attempt+1} failed: {e}")
        
        return PromptResult(
            answer="处理失败，请稍后重试",
            confidence=0.0,
            reasoning="Max retries exceeded",
            task_type=task_type,
            retry_count=self.max_retries
        )


# 使用示例
router = PromptRouter()

# 数学题自动走 CoT
r1 = router.process("计算 1+2+3+...+100 的和是多少？")
print(f"任务类型：{r1.task_type.value}")
print(f"置信度：{r1.confidence}")
print(f"策略：{r1.reasoning}")

# 结构化提取
schema = {
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "age": {"type": "integer"},
        "city": {"type": "string"}
    },
    "required": ["name", "age", "city"]
}
r2 = router.process(
    "从以下文本提取信息：小明今年 25 岁，住在北京市。",
    schema=schema
)
print(f"\n提取结果：{r2.answer}")