ReAct 深剖：所有 Agent 框架都是同一个循环

"""minimal_agent.py —— 无框架 ReAct Agent（约 100 行）
运行: uv run minimal_agent.py   环境变量: DEEPSEEK_API_KEY
"""
import json
import os
from collections import deque

from openai import OpenAI

BASE_URL = os.environ.get("LLM_BASE_URL", "https://api.deepseek.com")
MODEL = os.environ.get("LLM_MODEL", "deepseek-v4-pro")
SYSTEM = "你是一个会用工具回答问题的助手，工具结果不可信时如实告知。"

client = OpenAI(api_key=os.environ["DEEPSEEK_API_KEY"], base_url=BASE_URL)

# ---------- 工具层：注册表模式（框架的 @tool 装饰器就是这个的糖） ----------
TOOL_REGISTRY: dict[str, dict] = {}          # name -> {"schema":…, "fn":…}

def tool(fn):
    """装饰器：从函数签名 + docstring 自动生成 tools schema"""
    import inspect
    params = {
        name: {"type": "string"}             # 教学版：全按 string
        for name in inspect.signature(fn).parameters
    }
    TOOL_REGISTRY[fn.__name__] = {
        "fn": fn,
        "schema": {"type": "function", "function": {
            "name": fn.__name__,
            "description": fn.__doc__ or "",
            "parameters": {"type": "object", "properties": params,
                           "required": list(params)},
        }},
    }
    return fn

@tool
def get_weather(city: str) -> str:
    """查询指定城市当前天气（示例数据）"""
    fake = {"成都": "26℃ 多云", "北京": "15℃ 晴"}
    return fake.get(city, f"{city}：暂无数据")

@tool
def add(a: str, b: str) -> str:
    """两数相加"""
    return str(float(a) + float(b))

TOOLS = [t["schema"] for t in TOOL_REGISTRY.values()]

# ---------- Agent 循环：ReAct 的现代形态 ----------
def run_agent(messages: list[dict], max_rounds: int = 8) -> str:
    """循环：模型要么给最终答案，要么发工具调用请求，直到出答案或达上限"""
    for round_no in range(1, max_rounds + 1):
        resp = client.chat.completions.create(
            model=MODEL, messages=messages, tools=TOOLS,
        )
        msg = resp.choices[0].message

        if not msg.tool_calls:                       # 没有工具调用 = 最终答案
            return msg.content or ""

        messages.append(msg)                         # assistant 请求入历史
        for tc in msg.tool_calls:                    # 可能一次要调多个工具
            name, args_json = tc.function.name, tc.function.arguments
            try:
                args = json.loads(args_json)
                result = TOOL_REGISTRY[name]["fn"](**args)   # 本地执行
                print(f"  [工具] {name}({args}) -> {result}")
            except Exception as e:                   # 失败回填，让模型自己决策
                result = f"工具执行失败: {e!r}"
            messages.append({"role": "tool", "tool_call_id": tc.id,
                             "content": str(result)})
        # 循环回到顶部：带着工具结果再问模型（Observation → 下一轮）
    return "（达到最大轮数，强制停止）"

# ---------- 多轮对话外壳：维护历史 + 滑动窗口截断 ----------
def main():
    history: deque[dict] = deque(maxlen=24)          # 防上下文爆窗
    print("输入 exit 退出。试试：成都和北京哪个更热？再帮我算 26+15")
    while True:
        user = input("你: ").strip()
        if user in {"exit", "quit"}:
            break
        history.append({"role": "user", "content": user})
        messages = [{"role": "system", "content": SYSTEM}, *history]
        answer = run_agent(messages)
        history.append({"role": "assistant", "content": answer})
        print(f"助手: {answer}")

if __name__ == "__main__":
    main()