AI 智能体工作流：从零构建可复用的 ReAct 代理系统

Thought: 思考当前需要做什么
Action: 选择一个工具并传入参数
Observation: 观察工具返回的结果
... (重复上述步骤直到完成)
Final Answer: 给出最终答案

pip install openai requests python-dotenv

OPENAI_API_KEY=your_key_here
OPENAI_BASE_URL=https://api.openai.com/v1  # 可替换为代理或本地端点

import requests
import json
import re

# 工具注册字典
TOOLS = {}

def register_tool(func):
    """装饰器：注册工具"""
    TOOLS[func.__name__] = func
    return func

@register_tool
def get_weather(city: str) -> str:
    """查询城市当前天气（使用 wttr.in 免费 API）"""
    try:
        url = f"https://wttr.in/{city}?format=%C+%t"
        resp = requests.get(url, timeout=5)
        return resp.text.strip()
    except Exception as e:
        return f"天气查询失败: {str(e)}"

@register_tool
def calculator(expression: str) -> str:
    """安全计算数学表达式，例如 '2 + 3 * 4'"""
    # 移除危险字符，只允许数字和运算符
    sanitized = re.sub(r'[^0-9+\-*/().]', '', expression)
    try:
        result = eval(sanitized, {"__builtins__": {}})
        return str(result)
    except Exception as e:
        return f"计算错误: {str(e)}"

@register_tool
def search(query: str) -> str:
    """模拟搜索，返回固定结果（实际可接入 SerpAPI 或 Bing）"""
    # 这里模拟一些常见问题，实际可替换为真实搜索 API
    mock_db = {
        "北京人口": "北京常住人口约 2185 万（2023 年）",
        "Python 版本": "最新稳定版为 Python 3.12",
        "AI 智能体": "AI 智能体是一种能够自主感知环境、做出决策并执行动作的实体。"
    }
    for key, value in mock_db.items():
        if key in query:
            return value
    return f"未找到关于 '{query}' 的准确信息，建议进一步核实。"

SYSTEM_PROMPT = """你是一个智能助理，能够使用工具解决用户问题。
你的回答必须严格遵循以下格式：

Thought: 你当前的想法，分析需要做什么。
Action: 你要调用的工具名称，必须是 [{tool_names}] 之一。
Action Input: 工具的输入参数（JSON 字符串）。

如果已经获得足够信息，则输出：
Thought: 我已经得到最终答案。
Final Answer: 对用户的最终回复。

注意：每次只输出一个 Thought/Action/Action Input 组合，不要输出多余内容。
工具列表及说明：
{tool_descriptions}
"""

def build_prompt(tools, messages):
    tool_names = ", ".join(tools.keys())
    tool_descriptions = "\n".join(
        [f"- {name}: {func.__doc__}" for name, func in tools.items()]
    )
    system = SYSTEM_PROMPT.format(
        tool_names=tool_names,
        tool_descriptions=tool_descriptions
    )
    # 将消息列表组装成 LLM 所需格式
    prompt_messages = [{"role": "system", "content": system}] + messages
    return prompt_messages

import os
import json
from openai import OpenAI

client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL")
)

def run_agent(user_query, max_steps=5):
    messages = [{"role": "user", "content": user_query}]
    step = 0
    while step < max_steps:
        step += 1
        print(f"\n--- Step {step} ---")
        
        # 调用 LLM
        prompt = build_prompt(TOOLS, messages)
        response = client.chat.completions.create(
            model="gpt-4o-mini",  # 可替换为其他模型
            messages=prompt,
            temperature=0.2
        )
        output = response.choices[0].message.content
        print(f"LLM Output:\n{output}")
        
        # 解析输出
        if "Final Answer" in output:
            # 提取最终答案
            final_match = re.search(r"Final Answer:\s*(.+)", output, re.DOTALL)
            if final_match:
                return final_match.group(1).strip()
            else:
                return "无法提取最终答案。"
        
        # 尝试提取 Action 和 Action Input
        action_match = re.search(r"Action:\s*(\w+)", output)
        input_match = re.search(r"Action Input:\s*(\{.+\})", output, re.DOTALL)
        
        if not action_match or not input_match:
            # 如果格式不对，要求 LLM 重新输出
            messages.append({"role": "assistant", "content": output})
            messages.append({"role": "user", "content": "请按照指定格式输出，包含 Action 和 Action Input。"})
            continue
        
        action_name = action_match.group(1)
        try:
            action_input = json.loads(input_match.group(1))
        except json.JSONDecodeError:
            messages.append({"role": "assistant", "content": output})
            messages.append({"role": "user", "content": "Action Input 必须是合法的 JSON 字符串。"})
            continue
        
        # 调用工具
        if action_name in TOOLS:
            tool_func = TOOLS[action_name]
            try:
                if isinstance(action_input, dict):
                    result = tool_func(**action_input)  # 传递参数
                else:
                    result = tool_func(action_input)
            except Exception as e:
                result = f"工具调用异常: {str(e)}"
        else:
            result = f"未知工具: {action_name}"
        
        print(f"Tool Result: {result}")
        
        # 将观察结果加入历史
        messages.append({"role": "assistant", "content": output})
        messages.append({"role": "user", "content": f"Observation: {result}"})
    
    return "超过最大迭代步数，未能完成任务。"

if __name__ == "__main__":
    query = "明天北京天气怎么样？顺便帮我算一下 (25 + 30) * 2 等于多少。"
    answer = run_agent(query, max_steps=4)
    print("\n=== Final Answer ===")
    print(answer)

--- Step 1 ---
LLM Output:
Thought: 用户需要查询北京明天的天气，并计算一个数学表达式。我需要分别调用天气和计算器工具。
Action: get_weather
Action Input: {"city": "北京"}
Tool Result: Partly cloudy +22°C

--- Step 2 ---
LLM Output:
Thought: 天气已获取，现在需要计算 (25+30)*2。
Action: calculator
Action Input: {"expression": "(25+30)*2"}
Tool Result: 110

--- Step 3 ---
LLM Output:
Thought: 我已经获得了天气和计算结果，可以给出最终答案。
Final Answer: 明天北京天气为 partly cloudy，气温约 22°C。您计算的 (25+30)*2 等于 110。

=== Final Answer ===
明天北京天气为 partly cloudy，气温约 22°C。您计算的 (25+30)*2 等于 110。

workflow:
  - step: query_weather
    tool: get_weather
    params: { city: input.city }
    next: calculate
  - step: calculate
    tool: calculator
    params: { expression: input.expr }
    next: aggregate
  - step: aggregate
    prompt: "整合天气和计算结果，生成友好回复"