ReAct 模式与主循环

Question: 特斯拉 2024 年交付了多少辆车？
Thought: 我需要搜索特斯拉 2024 年的交付数据
Action: search[特斯拉 2024 交付量]
Observation:                                        ← 程序停在这里！

from openai import OpenAI

client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")

# 模拟一个可检索的知识库（第 9 章换成真实搜索 API）
KB = {
    "特斯拉 2024 交付量": "特斯拉 2024 年全球交付约 179 万辆",
    "比亚迪 2024 交付量": "比亚迪 2024 年销量约 427 万辆",
    "qwen3 参数": "Qwen3 系列从 0.6B 到 235B 共 8 个规模",
}

def search(query: str) -> str:
    for key, val in KB.items():
        if key in query or query in key:
            return val
    return "未找到相关信息，建议换一个关键词"

import re

TOOLS_DOC = """可用工具：
- search[查询词]：在知识库中检索信息"""

PROMPT_TPL = """{tools_doc}

严格按以下格式逐步回答问题（一次只输出一个 Thought 或 Action，输出到 Action 后停下）：

Question: {question}
"""

def react(question: str, max_steps: int = 6) -> str:
    transcript = ""      # 累积的 Thought / Action / Observation 记录

    for step in range(max_steps):
        prompt = PROMPT_TPL.format(tools_doc=TOOLS_DOC, question=question)
        prompt += transcript + "/no_think"      # qwen3 的软开关：关闭内置思考模式
        resp = client.chat.completions.create(
            model="qwen3:4b",
            messages=[{"role": "user", "content": prompt}],
            stop=["Observation:"],              # 第 4 章的 stop 序列：生成到此为止
            temperature=0,
        )
        output = resp.choices[0].message.content.strip()
        transcript += f"{output}\nObservation:"  # 拼回上下文，等待填入观察
        print(f"--- step {step + 1} ---\n{output}\n")

        # 终止条件 1：给出最终答案
        m = re.search(r"Final Answer:\s*(.+)", output, re.S)
        if m:
            return m.group(1).strip()

        # 终止条件 2：发起工具调用
        m = re.search(r"Action:\s*(\w+)\[(.+?)\]", output, re.S)
        if not m:
            # 模型没按格式来：把观察置为提示，请它重新遵守格式
            transcript += "（格式错误：请输出 Thought: 或 Action: 工具[参数]）\n"
            continue

        tool_name, arg = m.group(1), m.group(2).strip()
        if tool_name == "search":
            observation = search(arg)
        else:
            observation = f"错误：不存在工具 {tool_name}"
        transcript += f" {observation}\n"

    return "（达到最大步数限制）"

print("答案:", react("比亚迪和特斯拉 2024 年谁卖得多？多多少？"))

--- step 1 ---
Thought: 我需要先查比亚迪 2024 年的销量
Action: search[比亚迪 2024 交付量]
Observation: 比亚迪 2024 年销量约 427 万辆

--- step 2 ---
Thought: 拿到比亚迪的数据了，接着查特斯拉
Action: search[特斯拉 2024 交付量]
Observation: 特斯拉 2024 年全球交付约 179 万辆

--- step 3 ---
Thought: 两边数据都有了：427 - 179 = 248，比亚迪多
Final Answer: 比亚迪 2024 年销量约 427 万辆比特斯拉的约 179 万辆多，多了约 248 万辆。

def agent_loop(user_input, registry, messages):
    messages.append({"role": "user", "content": user_input})
    while True:
        resp = client.chat.completions.create(
            model="qwen3:4b", messages=messages, tools=registry.schemas(),
        )
        msg = resp.choices[0].message
        messages.append(msg)

        if not msg.tool_calls:              # 终止：模型不再要工具 → 最终回答
            return msg.content

        for tc in msg.tool_calls:           # 执行所有调用并回传
            result = registry.dispatch(tc.function.name, tc.function.arguments)
            messages.append({"role": "tool", "tool_call_id": tc.id, "content": result})

形态一（鬼打墙）：
  search["qwen3 发布时间"] → 结果不满意 → search["qwen3 发布时间"] → ...

形态二（打乒乓）：
  read_file[a.txt] → read_file[b.txt] → read_file[a.txt] → read_file[b.txt] → ...

def is_stuck(recent_calls: list[tuple], window: int = 3) -> bool:
    """最近 window 次调用完全相同时判定卡死"""
    if len(recent_calls) < window:
        return False
    return len(set(recent_calls[-window:])) == 1

# 事件类型常量
class Ev:
    ROUND_START = "round_start"     # 一轮开始
    TOOL_CALL    = "tool_call"       # 模型请求调用工具
    TOOL_RESULT = "tool_result"     # 工具执行完毕
    FINISH       = "finish"         # 自然终止
    ABORT        = "abort"          # 异常终止（超轮次/超预算）
    NUDGE        = "nudge"          # 检测到卡死，注入纠偏提示

from openai import OpenAI

client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")

class Agent:
    def __init__(self, system: str, registry, model: str = "qwen3:4b",
                 max_rounds: int = 8, token_budget: int = 20_000):
        self.registry = registry
        self.model = model
        self.max_rounds = max_rounds
        self.token_budget = token_budget
        self.messages = [{"role": "system", "content": system}]
        self.tokens_used = 0
        self.recent_calls: list[tuple] = []       # 重复检测用
        self.listeners = []                        # 事件监听器

    # ---------- 事件系统 ----------
    def on(self, listener):
        """注册监听器：fn(event: str, data: dict)"""
        self.listeners.append(listener)

    def emit(self, event: str, **data):
        for fn in self.listeners:
            fn(event, data)

    # ---------- 主入口 ----------
    def run(self, user_input: str) -> str:
        self.messages.append({"role": "user", "content": user_input})

        for round_no in range(1, self.max_rounds + 1):
            # 闸门 3：token 预算
            if self.tokens_used > self.token_budget:
                self.emit(Ev.ABORT, reason="token_budget_exceeded")
                return "（已达 token 预算上限，任务中止）"

            self.emit(Ev.ROUND_START, round=round_no, tokens=self.tokens_used)
            resp = client.chat.completions.create(
                model=self.model,
                messages=self.messages,
                tools=self.registry.schemas(),
            )
            msg = resp.choices[0].message
            self.messages.append(msg)
            self.tokens_used += resp.usage.total_tokens

            # 自然终止
            if not msg.tool_calls:
                self.emit(Ev.FINISH, round=round_no, tokens=self.tokens_used)
                return msg.content

            # 执行本轮所有调用
            for tc in msg.tool_calls:
                signature = (tc.function.name, tc.function.arguments)
                self.recent_calls.append(signature)
                self.emit(Ev.TOOL_CALL, **{"call": signature})

                result = self.registry.dispatch(*signature)
                self.emit(Ev.TOOL_RESULT, **{"call": signature, "result": result[:200]})
                self.messages.append({
                    "role": "tool",
                    "tool_call_id": tc.id,
                    "content": result,
                })

            # 闸门 4：重复检测（最近 3 次完全相同）
            if is_stuck(self.recent_calls):
                self.emit(Ev.NUDGE, calls=self.recent_calls[-3:])
                self.messages.append({                # 告知模型并给出最后一次机会
                    "role": "user",
                    "content": "检测到你连续多次发起完全相同的调用。请改变策略或给出最终答案。",
                })

        # 闸门 2：轮次上限
        self.emit(Ev.ABORT, reason="max_rounds_exceeded")
        return "（已达最大轮次上限，任务未完成）"

agent = Agent(
    system="你是一个文件助手，可以读写本地文件、做精确计算。",
    registry=registry,          # 第 7 章的 Registry 实例
)

def logger(event, data):
    if event == Ev.TOOL_CALL:
        print(f"  ⚙ 调用 {data['call'][0]}({data['call'][1][:60]})")
    elif event == Ev.TOOL_RESULT:
        print(f"  ↳ {data['result'][:80]}")
    elif event == Ev.FINISH:
        print(f"  ✔ 完成，共 {data['round']} 轮，消耗 {data['tokens']} tokens")
    elif event == Ev.ABORT:
        print(f"  ✖ 中止：{data['reason']}")

agent.on(logger)
print(agent.run("帮我算一下 (128+256)*3 的结果，写入 result.txt，然后读回来确认"))

  ⚙ 调用 calculate({"expression": "(128+256)*3"})
  ↳ 1152
  ⚙ 调用 write_file({"path": "result.txt", "content": "1152"})
  ↳ 已写入 result.txt（4 字符）
  ⚙ 调用 read_file({"path": "result.txt"})
  ↳ 1152
  ✔ 完成，共 4 轮，消耗 2841 tokens