推理增强技术工程化：CoT/ToT/ReAct

# cot.py — Zero-shot 与 Few-shot CoT
ZERO_SHOT_COT = "请一步一步推理，最后给出答案。"

FEW_SHOT_COT = [
    ("如果 5 个工人 5 天能铺 5 里路，10 个工人 10 天能铺多少里？",
     "5 个工人 5 天铺 5 里 → 每个工人每天铺 5/25 = 0.2 里。"
     "10 个工人 10 天铺 0.2 × 10 × 10 = 20 里。答案是 20 里。"),
]

def cot_query(question: str, use_few_shot=False) -> str:
    if use_few_shot:
        return few_shot_prompt(FEW_SHOT_COT, question)
    return question + "\n" + ZERO_SHOT_COT

def parse_cot_output(text: str, answer_pattern=r"答案是\s*(.+?)[。\n]") -> dict:
    """从 CoT 输出中提取推理与答案。"""
    match = re.search(answer_pattern, text)
    return {
        "reasoning": text[:match.start()] if match else text,
        "answer": match.group(1) if match else None,
    }

def should_use_cot(query: str, classifier) -> bool:
    """简单题不要 CoT，复杂题才用——控成本。"""
    return classifier(query)["complexity"] != "simple"

# self_consistency.py — 自洽性采样
from collections import Counter

def self_consistency(question: str, n_samples=5, temperature=0.7) -> dict:
    """采样 N 条 CoT 路径，对最终答案投票。"""
    answers = []
    for _ in range(n_samples):
        text = cot_query(question)
        parsed = parse_cot_output(text)
        answers.append(parsed["answer"])
    votes = Counter(a for a in answers if a is not None)
    final, count = votes.most_common(1)[0]
    return {
        "final": final,
        "votes": dict(votes),
        "consensus": count / len(answers),   # 共识度 = 置信度信号
    }

def self_consistency_with_escalation(question, judge, upgrade_fn):
    """共识度低 → 升级模型再采。"""
    result = self_consistency(question, n_samples=5)
    if result["consensus"] >= 0.8:
        return result["final"], "small"
    result2 = self_consistency(question, n_samples=5, model="flagship")
    return result2["final"], "flagship"

# tot.py — 思维树框架
import heapq

def tot_solve(problem, generator, evaluator,
              max_depth=4, branch=3, top_k=2) -> list[dict]:
    """
    generator: 输入状态 → 产出若干下一步候选
    evaluator: 输入状态 → 打分（评估该分支质量）
    搜索：每层保留 top_k 高分分支，树状扩展。
    """
    frontier = [{"state": problem, "depth": 0, "path": []}]
    solutions = []
    for _ in range(max_depth):
        next_frontier = []
        for node in frontier:
            candidates = generator(node["state"])     # 分支
            for cand in candidates:
                score = evaluator(cand)               # 评估
                next_frontier.append({
                    "state": cand, "score": score,
                    "depth": node["depth"] + 1,
                    "path": node["path"] + [cand],
                })
        # 剪枝：只保留 top_k 高分
        frontier = heapq.nlargest(top_k, next_frontier, key=lambda n: n["score"])
        if any(is_solution(n["state"]) for n in frontier):
            break
    return frontier

def tot_budget(max_depth, branch, top_k) -> int:
    """估算 ToT 的 LLM 调用次数上限。"""
    calls = 0
    frontier = 1
    for _ in range(max_depth):
        calls += frontier * branch        # 生成 + 评估
        frontier = min(top_k, frontier * branch)
    return calls

# react.py — ReAct 循环引擎
async def react_loop(query: str, tools: dict, max_steps=5) -> str:
    """Thought → Action → Observation 循环，直到给出 Answer。"""
    history = [f"Question: {query}"]
    for step in range(max_steps):
        step_out = await model.react_step(history)   # 输出 Thought + Action
        history.append(step_out)

        action = parse_action(step_out)              # 解析 Action
        if action is None:
            answer = parse_answer(step_out)
            if answer:
                return answer
            continue

        result = run_tool(tools, action)             # 执行工具
        history.append(f"Observation: {result}")
    return "达到最大步数，未得到答案"

import re

ACTION_RE = re.compile(r"Action:\s*(\w+)\s*\((.+?)\)")

def parse_react_output(text: str) -> dict:
    """解析 ReAct 输出：可能含 Thought、Action、或最终 Answer。"""
    thought = re.search(r"Thought:\s*(.+?)(?=Action|Answer)", text, re.DOTALL)
    action = ACTION_RE.search(text)
    answer = re.search(r"Answer:\s*(.+)", text, re.DOTALL)
    return {
        "thought": thought.group(1).strip() if thought else None,
        "action": {"name": action.group(1), "args": action.group(2)}
                  if action else None,
        "answer": answer.group(1).strip() if answer else None,
    }

def safe_run_tool(tools, action, max_steps):
    """工具执行前的白名单校验与步数守卫。"""
    if action["name"] not in tools:
        return "错误：未知工具，请停止"
    return tools[action["name"]](action["args"])

REASONING_SCHEMA = {
    "type": "object",
    "properties": {
        "chain": {"type": "array",
                  "items": {"type": "object",
                            "properties": {"step": {"type": "string"}}}},
        "answer": {"type": "string"},
        "confidence": {"type": "number"},
    },
    "required": ["chain", "answer"],
}

def structured_reason(query: str) -> dict:
    """结构化推理输出：步骤数组 + 答案 + 置信度。"""
    resp = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": query}],
        response_format={"type": "json_schema", "schema": REASONING_SCHEMA},
    )
    return json.loads(resp.choices[0].message.content)

def reasoning_with_checkpoints(query, check_fn, max_steps=5):
    """每步自检：不通过则终止并降级。"""
    steps = []
    for i in range(max_steps):
        step = model.next_reasoning_step(query, steps)
        steps.append(step)
        if not check_fn(step):           # 该步逻辑校验失败
            return {"status": "aborted", "reason": f"step{i} 校验失败"}
    return {"status": "done", "steps": steps,
            "answer": model.finalize(query, steps)}

def evaluate_reasoning(queries, pipeline, golden) -> dict:
    """答案正确率 + 推理步骤质量抽样。"""
    acc = 0
    reasoning_samples = []
    for q, ref in zip(queries, golden):
        out = pipeline(q)
        acc += (out["answer"] == ref["answer"])
        reasoning_samples.append(judge_reasoning(out["chain"], ref))
    return {
        "accuracy": acc / len(queries),
        "reasoning_quality": mean(reasoning_samples),
    }

def route_reasoning(query: str, classifier) -> dict:
    """按任务类型选择推理策略。"""
    task = classifier(query)["task"]
    strategy = {
        "extraction":      lambda q: direct_answer(q),          # 不需要推理
        "qa":              lambda q: cot_query(q),              # CoT 即可
        "math":            lambda q: self_consistency(q, 5),    # 采样投票
        "planning":        lambda q: tot_solve(q, ...),         # ToT 搜索
        "tool_task":       lambda q: react_loop(q, tools),      # ReAct
    }[task]
    return strategy(query)

def trace_reasoning(query, strategy, token_counter):
    """记录推理链路：策略、步数、token、耗时。"""
    start = time.time()
    out = strategy(query)
    return {
        "strategy": strategy.__name__,
        "steps": len(out.get("chain", [])),
        "tokens": token_counter(query, out),
        "latency_ms": (time.time() - start) * 1000,
        "answer": out.get("answer"),
    }