循环范式

"""范式一 · 原生 function calling：与 ch02 完全相同——这就是重点。

运行：python native.py "1234 * 5678 等于多少？再开平方呢？"
"""
import os
import sys

from openai import OpenAI

from tools import registry

client = OpenAI(base_url=os.getenv("OPENAI_BASE_URL"), api_key=os.environ["OPENAI_API_KEY"])
MODEL = os.getenv("MODEL", "gpt-4o-mini")
MAX_STEPS = 10


def run(task: str) -> str:
    messages = [
        {"role": "system", "content": "你是一个会用工具的助手。用工具获取事实，得到最终答案后直接回答。"},
        {"role": "user", "content": task},
    ]
    for step in range(1, MAX_STEPS + 1):
        msg = client.chat.completions.create(model=MODEL, messages=messages, tools=registry.schemas()).choices[0].message
        if not msg.tool_calls:
            return msg.content
        messages.append(msg)
        for call in msg.tool_calls:  # 模型可能一次请求多个工具调用（并行工具调用）
            result = registry.call(call.function.name, call.function.arguments)
            print(f"  [step {step}] {call.function.name}({call.function.arguments}) -> {str(result)[:80]}")
            messages.append({"role": "tool", "tool_call_id": call.id, "content": result})
    return "（超出最大步数，强制停止）"


if __name__ == "__main__":
    print("最终答案:", run(sys.argv[1] if len(sys.argv) > 1 else "现在几点？顺便算一下 365*24。"))

"""范式二 · ReAct 文本协议：不依赖 function calling API，靠 prompt 约定 + 正则解析。

运行：python react_text.py "1234 * 5678 等于多少？再开平方呢？"
"""
import os
import re
import sys

from openai import OpenAI

from tools import registry

client = OpenAI(base_url=os.getenv("OPENAI_BASE_URL"), api_key=os.environ["OPENAI_API_KEY"])
MODEL = os.getenv("MODEL", "gpt-4o-mini")
MAX_STEPS = 10

SYSTEM = """尽你所能回答问题。你可以使用以下工具：

{tools}

严格按照如下格式输出（一次只走一步）：

Thought: 思考接下来该做什么
Action: 工具名，必须是 [{names}] 之一
Action Input: 工具的输入参数，JSON 格式
Observation: 工具返回的结果（由系统填写，你不要生成这一行）
...（Thought/Action/Action Input/Observation 可以重复多次）
Thought: 我已经知道最终答案了
Final Answer: 对问题的最终回答"""

ACTION_RE = re.compile(r"Action:\s*(.+?)\s*\nAction Input:\s*(\{.*?\})", re.S)
FINAL_RE = re.compile(r"Final Answer:\s*(.*)", re.S)


def run(task: str) -> str:
    schemas = registry.schemas()
    system = SYSTEM.format(
        tools="\n".join(f"- {s['function']['name']}: {s['function']['description']}" for s in schemas),
        names=", ".join(s["function"]["name"] for s in schemas),
    )
    scratchpad = f"Question: {task}\n"
    for _ in range(MAX_STEPS):
        text = client.chat.completions.create(
            model=MODEL,
            messages=[{"role": "system", "content": system}, {"role": "user", "content": scratchpad}],
            stop=["Observation:"],  # 关键：拦住模型，不让它自己编造观察结果
        ).choices[0].message.content
        scratchpad += text
        if m := FINAL_RE.search(text):
            return m.group(1).strip()
        if m := ACTION_RE.search(text):
            obs = registry.call(m.group(1).strip(), m.group(2))
            print(f"  [{m.group(1).strip()}] {m.group(2)} -> {str(obs)[:80]}")
            scratchpad += f"\nObservation: {obs}\n"
        else:  # 格式漂移：文本协议的最大痛点（本章挑战会改进这里）
            scratchpad += "\nObservation: 输出格式不合法，请严格按 Thought/Action/Action Input 或 Final Answer 输出。\n"
    return "（超出最大步数，强制停止）"


if __name__ == "__main__":
    print("最终答案:", run(sys.argv[1] if len(sys.argv) > 1 else "1234 * 5678 等于多少？再开平方呢？"))

"""范式三 · CodeAct：模型的"行动"不是 JSON 工具调用，而是直接写 Python 代码。

运行：python codeact.py "统计 1 到 100 中所有质数的和"
"""
import os
import re
import subprocess
import sys
import tempfile

from openai import OpenAI

client = OpenAI(base_url=os.getenv("OPENAI_BASE_URL"), api_key=os.environ["OPENAI_API_KEY"])
MODEL = os.getenv("MODEL", "gpt-4o-mini")
MAX_STEPS = 8
TIMEOUT = 15  # 秒；沙箱的第一道防线

SYSTEM = """你通过编写 Python 代码来完成任务，一轮写一段。
- 用 ```python 代码块输出代码；用 print() 打印你需要观察的中间结果。
- 代码在独立进程里执行，轮与轮之间不共享变量；需要延续的状态请重新计算或写入文件。
- 得到最终答案后，不要再写代码块，直接输出：FINAL: 你的答案"""

CODE_RE = re.compile(r"```python\n(.*?)```", re.S)


def run_sandboxed(code: str) -> str:
    """子进程 + 超时 = 最简"沙箱"。它挡得住死循环，挡不住文件读写——真正的隔离见 runtime-execution。"""
    with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False, encoding="utf-8") as f:
        f.write(code)
        path = f.name
    try:
        r = subprocess.run([sys.executable, path], capture_output=True, text=True, timeout=TIMEOUT)
        return (r.stdout + r.stderr).strip() or "（无输出——记得用 print()）"
    except subprocess.TimeoutExpired:
        return f"执行超时（>{TIMEOUT}s），代码可能死循环了"
    finally:
        os.unlink(path)


def run(task: str) -> str:
    messages = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": task}]
    for _ in range(MAX_STEPS):
        text = client.chat.completions.create(model=MODEL, messages=messages).choices[0].message.content
        messages.append({"role": "assistant", "content": text})
        if "FINAL:" in text and not CODE_RE.search(text):
            return text.split("FINAL:", 1)[1].strip()
        if m := CODE_RE.search(text):
            obs = run_sandboxed(m.group(1))
            print(f"  [exec] {len(m.group(1))} 字符代码 -> {obs[:120]}")
            messages.append({"role": "user", "content": f"执行结果：\n{obs}"})
        else:
            messages.append({"role": "user", "content": "请输出 ```python 代码块，或用 FINAL: 给出最终答案。"})
    return "（超出最大步数，强制停止）"


if __name__ == "__main__":
    print("最终答案:", run(sys.argv[1] if len(sys.argv) > 1 else "统计 1 到 100 中所有质数的和"))