W1 · Prompt 四模式：CoT / Few-shot / ReAct / ToT

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

import os
from openai import OpenAI

client = OpenAI(api_key=os.getenv("DEEPSEEK_API_KEY"),
                base_url="https://api.deepseek.com")


def get_response(messages, **kwargs):
    """统一请求入口：流式接收 + 双通道收集（思考 + 正文）。
    注意：如果生成了正式 content，会把 assistant 消息存回 messages（见多轮对话）。"""
    response = client.chat.completions.create(
        model=kwargs.get("model", "deepseek-v4-flash"),
        messages=messages,
        stream=True,                                  # 流式：一截一截收
        reasoning_effort=kwargs.get("reasoning_effort", "low"),
        max_tokens=kwargs.get("max_tokens", 500),
    )
    reasoning_content, content = "", ""
    for chunk in response:
        delta = chunk.choices[0].delta
        if delta.reasoning_content:                   # 草稿纸（思考）
            reasoning_content += delta.reasoning_content
        if delta.content:                             # 正式稿（正文）
            content += delta.content
    if content:
        messages.append({"role": "assistant", "content": content})   # 存回历史
        return content
    return reasoning_content                          # 纯思考题兜底

1
2
3
4
5
6
7

def cot_prompt(question, system_prompt="你是一个逻辑缜密，思维严谨的推理助手"):
    """CoT：要求一步步思考再回答"""
    u = f"请一步步思考，再给出答案。\n问题:{question}"
    return [
        {"role": "system", "content": system_prompt},   # 人设（规则）
        {"role": "user", "content": u},                 # 任务（要求一步步想）
    ]

1
2
3
4
5
6
7

  小明有 3 个苹果，又买了 2 打（12 个/打），分给 4 个朋友，每人几个？

  ❌ 直接答：5 个            ← 跳步：忘了把"2 打"换算成 24
  ✅ 一步步想：
     1) 3 + 2×12 = 27 个
     2) 27 ÷ 4 = 6.75
     3) 每人 6.75 个

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

DEFAULT_EXAMPLES = [
    {"question": "我明天下午3点约了张医生复诊",
     "answer": "时间:明天下午3点;人物:张医生;事项:复诊"},
    {"question": "周五晚上和李总在望江楼吃饭",
     "answer": "时间:周五晚上;人物:李总;事项:吃饭"},
]

def few_shot_prompt(question, examples=DEFAULT_EXAMPLES,
                    system_prompt="你是擅长信息抽取的助手"):
    """Few-shot：先给 2 个问答范例，再问真问题。"""
    msgs = [{"role": "system", "content": system_prompt}]
    for ex in examples:
        msgs.append({"role": "user", "content": ex["question"]})    # 示例问题
        msgs.append({"role": "assistant", "content": ex["answer"]})  # 示例答案
    msgs.append({"role": "user", "content": question})               # 真问题
    return msgs

1
2
3
4
5
6

  用规则描述：  "判断用户意图，可能是任务、闲聊或情绪表达……"   ← 说不清
  用例子示范：
    user: "帮我订明天去上海的机票"    assistant: 任务
    user: "今天好累啊"                assistant: 情绪
    user: "讲个笑话吧"                assistant: 闲聊
    user: "提醒我晚上买牛奶"          → 模型自然会模仿成: 任务

1
2
3
4
5

# ❌ 错误：把示例塞进一段 user 话里
{"role": "user", "content": "示例：Q:... A:... 示例2：Q:... A:... 现在回答：..."}

# ✅ 正确：示例拆成 user/assistant 交替（见 few_shot_prompt）
[user 示例1问题] [assistant 示例1答案] [user 示例2问题] [assistant 示例2答案] [user 真问题]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

def get_response(messages, **kwargs):
    # ……（同第一节：流式收集 + 存回 assistant）……

# system 定义人设：Agent 老师
messages = [{"role": "system", "content":
    "你是一个耐心 Agent 老师，擅长用通俗的语言解释学生提到的问题。"
    "请根据问题一步步分析学生当前的学习阶段，给出最适合当前阶段的回答。"}]

while True:
    user_input = input("User: ")
    if user_input.lower() in ["exit", "quit"]:
        break
    messages.append({"role": "user", "content": user_input})   # 本轮问题入队
    response = get_response(messages)                          # 内部会把 assistant 存回
    print("Agent导师:", response)

1
2
3
4
5

  第 1 轮前：  [system, user₁]
  第 1 轮后：  [system, user₁, assistant₁]
  第 2 轮前：  [system, user₁, assistant₁, user₂]
  第 2 轮后：  [system, user₁, assistant₁, user₂, assistant₂]
  …… messages 每轮 +2，模型每轮都"记得"前面聊过什么

1
2
3
4
5

Thought: 我先想清楚要做什么
Action: 调哪个工具
Action Input: 传什么参数
Observation: 工具返回了什么
Answer: 最终答案

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

def react_prompt(question, system_prompt="你是多种能力的小助手", tools="搜索引擎、计算器"):
    """ReAct 文本版：让模型按 Thought/Action/Observation/Answer 格式回答。
    注意：这里 Observation 是模型自己"编"的——它还没真工具可用。"""
    u = (f"可用工具:{tools}。请严格按格式回答：\n"
         f"Thought: 先根据用户问题想清楚要干什么\n"
         f"Action: 调用哪个工具更适合该问题\n"
         f"Action Input: 参数\n"
         f"Observation: 工具返回结果\n"
         f"Answer: 最终答案\n\n问题:{question}")
    return [{"role": "system", "content": system_prompt},
            {"role": "user", "content": u}]

1
2
3
4
5

  Thought: 用户问天气，需要调用天气查询工具
  Action: get_weather
  Action Input: {"city": "北京"}
  Observation: 北京今天晴，26℃          ← ⚠️ 它自己编的！没人真查
  Answer: 北京今天晴，26℃

1
2
3
4
5

  普通对话：  用户 → 模型 → 答案            （一次问答）
  ReAct：     用户 → 想一步 → 做一步 → 看结果 → 再想 → 答案   （循环）

  ReAct 第一次把"行动"和"观察"装进了回答格式里
  → 它差的就是"真能行动"——那就是 Agent

1
2
3
4
5

  发散：生成多条路径（多个方案）
    ↓
  评估：给每条路径打分（优缺点/风险）
    ↓
  收敛：选出最优路径（并给出实施步骤）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

def tot_prompt(system_prompt, question, n=3):
    """ToT 单调用版：把"发散→评估→收敛"全塞进一个问题里"""
    u = (f"请针对以下问题，提供{n}个不同的解决方案，"
         f"然后逐步分析每个解决方案的优缺点，最后给出最优方案，"
         f"并说明理由。\n\n问题：{question}")
    return [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": u},
    ]

# 使用：
msgs = tot_prompt("你是资深的 Agent 架构师，擅长多种角度对比方案", user_input)
response = get_response(msgs, max_tokens=1000)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

def ask(question, system_prompt="你是擅长多种角度对比方案的资深 Agent 架构师",
        max_tokens=500):
    """单次问答：组装 system + user 并调用"""
    msgs = [{"role": "system", "content": system_prompt},
            {"role": "user", "content": question}]
    return get_response(msgs, max_tokens=max_tokens)


def tot_solve(question, n=3):
    """真 ToT：三次调用，三个专家接力。每步的结果喂给下一步。"""
    # ① 发散：规划专家给 n 条思路
    schemes = ask(
        f"请针对以下问题给出{n}种不同解决思路，编号列出：\n{question}\n",
        system_prompt="你是擅长发散思维的规划专家",
        max_tokens=600,
    )
    # ② 评估：评审专家逐条打分（上一步的产出喂进来）
    scores = ask(
        f"请对以下 {n} 个方案逐一打分(1-10分),并各用一句话说明理由:\n{schemes}",
        system_prompt="你是严格的评审专家,打分要客观",
        max_tokens=500,
    )
    # ③ 收敛：决策专家选最优（评分喂进来）
    best = ask(
        f"综合以下评分,选出最优方案,并给出具体实施步骤:\n{scores}",
        system_prompt="你是决策专家,直接给最终选择",
        max_tokens=600,
    )
    return schemes, scores, best


# 使用：
s1, s2, s3 = tot_solve(user_input)
print("方案:", s1)
print("评审:", s2)
print("决策:", s3)

1
2

  question ──▶ [规划专家] ──schemes──▶ [评审专家] ──scores──▶ [决策专家] ──▶ best
                发散                   评估                   收敛

1
2
3
4
5
6

  ① 自动化     系统自己判断并调用，不用每次人工选
  ② 执行动作   路由选中的不只是"一段文字"，而是一整套执行流程
     —— ReAct 会启动"调 API → 执行工具 → 结果回填"的循环
     —— ToT 是三次调用的编排
     —— 这是手动粘贴永远做不到的
  ③ 可观测    能打日志、能批量测试、能调策略

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

# ── 模式注册表：名字 → prompt 构建函数 ──
BUILDERS = {
    "cot": cot_prompt,
    "few_shot": few_shot_prompt,
    "react": react_prompt,        # 文本版（Observation 还是编的，真工具见 FC 篇）
}


def auto_select(question: str) -> str:
    """让 LLM 判断这题适合哪种模式，返回模式名。
    规则写在 prompt 里；识别用关键词兜底（tot→react→few_shot→cot）。
    匹配顺序从长到短，避免 'cot' 误吞 'react'。"""
    content = (
        f"判断下面用户的问题适合哪种回答模式，输出：tot 或 react 或 few_shot 或 cot\n"
        f"规则:需要一步步推理/数学解题/公式推导的输出 cot;"
        f"固定模板化抽取信息的输出 few_shot;"
        f"需要调工具(天气/计算等)的输出 react;"
        f"开放性问题/多方案对比找最优的输出 tot\n"
        f"问题:{question}"
    )
    text = get_response([{"role": "user", "content": content}],
                        max_tokens=50).lower()
    for mode in ["tot", "react", "few_shot", "cot"]:   # 先长后短
        if mode in text:
            return mode
    print("[警告] 未识别到模式名,兜底 cot")
    return "cot"


def run_mode(mode: str, question: str, **kw):
    """按模式执行。tot 特判（要拼三段输出），其余走 BUILDERS。"""
    if mode == "tot":
        schemes, scores, best = tot_solve(question, n=kw.get("n", 3))
        return f"【方案】\n{schemes}\n\n【评审】\n{scores}\n\n【决策】\n{best}"
    if mode not in BUILDERS:
        raise ValueError(f"未知模式:{mode},可选 {list(BUILDERS) + ['tot']}")
    msgs = BUILDERS[mode](question, **kw)
    return get_response(msgs, **kw)


# ── 主循环：选模式 → 执行 → 输出（Agent 的第一版驾驶舱）──
while True:
    user_input = input("User: ")
    if user_input.lower() in ["exit", "quit"]:
        break
    mode = auto_select(user_input)
    print(f"[使用模式] {mode}")
    result = run_mode(mode, user_input)
    print("Agent导师:", result)

1
2

  以前：你写死一种方式 → 调用
  现在：模型看问题 → 自己选 cot/few_shot/react/tot → 执行

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

# ===== 三种模式一句话 =====
# CoT      要求一步步思考 → 适合数学/推理（本质：一条指令）
# Few-shot 给范例照葫芦画瓢 → 适合格式固定任务（本质：user/assistant 问答对）
# ReAct    想一步做一步看一步 → Agent 前身（Observation 要真工具，见 FC 篇）

# ===== 模式构建函数 =====
def cot_prompt(q, sys="你是逻辑缜密的推理助手"):
    return [{"role": "system", "content": sys},
            {"role": "user", "content": f"请一步步思考，再给出答案。\n问题:{q}"}]

def few_shot_prompt(q, examples=[("小明约了张医生", "时间:明天;人物:张医生")], sys="..."):
    msgs = [{"role": "system", "content": sys}]
    for question, answer in examples:
        msgs.append({"role": "user", "content": question})       # 一问
        msgs.append({"role": "assistant", "content": answer})     # 一答
    msgs.append({"role": "user", "content": q})                   # 真问题
    return msgs

def react_prompt(q, tools="搜索引擎、计算器"):
    u = (f"可用工具:{tools}。请严格按格式回答：\nThought:...\nAction:...\n"
         f"Action Input:...\nObservation:...\nAnswer:...\n\n问题:{q}")
    return [{"role": "system", "content": "你是多种能力的小助手"},
            {"role": "user", "content": u}]

# ===== ToT 两种做法 =====
# 单调用：tot_prompt 一发带走（发散→评估→收敛 混在一次回答）
# 多轮：  tot_solve 三连调（规划→评审→决策，每步产出喂下一步）→ 关键决策用这个

# ===== Prompt 及格线 =====
# 角色任务清晰 + 输入输出边界明确 + 可复用可测试（与技巧高级无关）

1
2
3
4
5
6

  ① 文字（prompt 内容）和代码（messages 组装）缺一不可
     —— API 只收带 role 的结构，Few-shot 的问答对必须代码拆
  ② 模式可以注册成表（BUILDERS），用路由（auto_select）选
     —— 代码路由 = 自动化 + 执行动作 + 可观测，手动粘贴做不到
  ③ ReAct 的 Observation 必须是真工具的真结果
     —— 文本版只是"演戏"，接真工具才是 Agent