04_react_loop.py

import os
import json
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

client = OpenAI(
    api_key=os.getenv("LLM_API_KEY"),
    base_url=os.getenv("LLM_BASE_URL")
)

# --- 工具定义区 ---

def calculate(expression: str) -> str:
    """执行数学表达式计算"""
    try:
        return str(eval(expression))
    except Exception as e:
        return f"计算出错: {str(e)}"

# 定义工具映射表：将模型返回的“函数名字符串”映射到真正的“函数对象”
# 这种写法是所有专业 Agent 框架的通用做法
available_tools = {
    "calculate": calculate
}

# --- 工具描述区 (JSON Schema) ---

tools = [
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "数学计算器，支持 + - * / 等运算。",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {"type": "string", "description": "数学表达式，如 '123 * 456'"}
                },
                "required": ["expression"]
            }
        }
    }
]

class ReActAgent:
    def __init__(self, system_prompt: str):
        self.history = [{"role": "system", "content": system_prompt}]

    def run(self, user_prompt: str) -> str:
        self.history.append({"role": "user", "content": user_prompt})
        
        step = 1
        while True:
            print(f"\n--- 🔄 第 {step} 轮推理 ---")
            
            response = client.chat.completions.create(
                model=os.getenv("LLM_MODEL_ID"),
                messages=self.history,
                tools=tools,
                tool_choice="auto"
            )
            
            msg = response.choices[0].message
            self.history.append(msg)
            
            if msg.content:
                print(f"🤔 模型思考: {msg.content}")
            
            if not msg.tool_calls:
                return msg.content
            
            for tool_call in msg.tool_calls:
                func_name = tool_call.function.name
                # 【动态调用关键点】：通过函数名字符串从字典中获取函数对象
                func_to_call = available_tools.get(func_name)
                
                # 解析参数
                args = json.loads(tool_call.function.arguments)
                
                if func_to_call:
                    print(f"🛠️  执行动作: 动态调用 {func_name}(**{args})")
                    # 使用 **args 自动匹配函数参数，实现通用化调用
                    observation = func_to_call(**args)
                else:
                    observation = f"错误：未找到工具 {func_name}"
                
                print(f"👁️  观察结果: {observation}")
                
                self.history.append({
                    "role": "tool",
                    "content": observation,
                    "tool_call_id": tool_call.id
                })
            
            step += 1

if __name__ == "__main__":
    print("=== 技术模块 04: ReAct 自主推理循环 ===")
    
    sys_msg = """你是一个具备数学运算能力的自主智能体。
你的思考过程必须遵循：
1. Thought: 思考当前需要做什么。
2. Action: 调用提供的工具。
3. 得到 Observation 后，继续思考下一步。
直到得出最终结论。"""

    agent = ReActAgent(system_prompt=sys_msg)
    result = agent.run("请先计算 152 乘以 48 等于多少，然后在那个结果的基础上加上 1000。")
    print(f"\n✅ 任务最终完成: {result}")