实现一个基于 API 调用的、最简单、最标准的 ReAct 智能体

get_weather()

f字符串 = 格式化字符串

作用：把变量直接塞进字符串里

例子：

1
2
3

name = "小明"
print(f"你好{name}")
# 输出：你好小明

requests库

官方库文档：快速入门 — Requests 2.32.3 文档 - Requests 中文

简单写一个get_weather()函数

import requests

def get_weather(city: str) -> str:
    url = f'https://wttr.in/{city}?format=j1'
    #加了format=j1才能返回json
    response = requests.get(url)
    #get获取页面返回的所有信息
    response.raise_for_status()
    #检查状态码是否为200
    print(response.json())
    
get_weather('北京')

json数据提取

{} 大括号代表一个 JSON 对象（对应 Python 字典），[] 中括号才代表 JSON 数组（对应 Python 列表）

{
  "nearest_area": [
    {
      "areaName": [{"value": "Beijing"}]
    }
  ],
  "current_condition": [      <-- 重点在这里
    {
      "weatherDesc": [        <-- 天气描述是数组
        {
          "value": "Thundery outbreaks in nearby"
        }
      ],
      "temp_C": "17",         <-- 摄氏温度
      "temp_F": "63",
      "windspeed": "12"
    }
  ],
  "request": [{}]
}

做一个简单的json数据清洗

return f"{city}: {data['current_condition'][0]['weatherDesc'][0]['value']}, {data['current_condition'][0]['temp_C']}°C"

这一块就是[current_condition][0]

{
      "weatherDesc": [        <-- 天气描述是数组
        {
          "value": "Thundery outbreaks in nearby"
        }
      ],
      "temp_C": "17",         <-- 摄氏温度
      "temp_F": "63",
      "windspeed": "12"
    }

输出北京: Thundery outbreaks in nearby, 17°C

异常处理

try:
    
except Exception:

Exception = 所有 “程序错误” 的总称

try 试一下

Exception 所有程序错误

except 抓到了就处理

加上异常处理后

import requests

def get_weather(city: str) -> str:
    url = f"https://wttr.in/{city}?format=j1"
    #print(url)
    try:
        response = requests.get(url)
        response.raise_for_status()
        data = response.json()
        #print(response.json())
        return f"{city}: {data['current_condition'][0]['weatherDesc'][0]['value']}, {data['current_condition'][0]['temp_C']}°C"
    except Exception:
        return '查询失败'

print(get_weather('北京'))

更精准的异常处理

except requests.exceptions.RequestException as e:
    return f"错误:查询天气时遇到网络问题 - {e}"

except (KeyError, IndexError) as e:
    return f"错误:解析天气数据失败 - {e}"

1. KeyError
JSON 里没有这个字段
比如：
data['abcdefg'] # 根本没有这个键 → 报 KeyError
2. IndexError
列表下标越界
比如：
data['current_condition'][999] # 没有第999个元素 → 报 IndexError
意思：
网通了，但数据拿不到 / 结构不对 → 走这里！

最终超级记忆口诀

RequestException = 网络 / 请求错
KeyError = 找不到键
IndexError = 下标越界
分开捕获 = 精准报错，好排查！

异常是如何接到的呢？

异常是在执行try内的代码时，如果执行 requests.get(url)时，python发现连不上网就自动抛出一个“RequestException” 错误，程序立刻停止往下走
直接跳到对应的 except 里

完整实现函数

import requests

def get_weather(city: str) -> str:
    """
    通过调用 wttr.in API 查询真实的天气信息。
    """
    # API端点，我们请求JSON格式的数据
    url = f"https://wttr.in/{city}?format=j1"
    
    try:
        # 发起网络请求
        response = requests.get(url)
        # 检查响应状态码是否为200 (成功)
        response.raise_for_status() 
        # 解析返回的JSON数据
        data = response.json()
        
        # 提取当前天气状况
        current_condition = data['current_condition'][0]
        weather_desc = current_condition['weatherDesc'][0]['value']
        temp_c = current_condition['temp_C']
        
        # 格式化成自然语言返回
        return f"{city}当前天气:{weather_desc}，气温{temp_c}摄氏度"
        
    except requests.exceptions.RequestException as e:
        # 处理网络错误
        return f"错误:查询天气时遇到网络问题 - {e}"
    except (KeyError, IndexError) as e:
        # 处理数据解析错误
        return f"错误:解析天气数据失败，可能是城市名称无效 - {e}"

get_attraction

为什么用Tavily

Welcome - Tavily Docs

喂给ai的技术文档docs.tavily.com/llms.txt

from tavily import TavilyClient

tavily_client = TavilyClient(api_key="tvly-YOUR_API_KEY")
response = tavily_client.search("Who is Leo Messi?")

print(response)

搜索天气是因为有稳定干净的json数据源，只需要简单清洗就能拿到我们想要的，但是像景点这些就可能会涉及各种爬虫，要模拟浏览器、带cookies等等，所以选择用Tavily，它是别人帮你做好的「专业搜索爬虫 + 清洗 + 结构化 API」

你发：

1	client.search("今天成都天气")

它直接返回 干净 JSON：

{
  "results": [
    {"title": "...", "url": "...", "content": "摘要..."},
    ...
  ]
}

省去了我们数据清洗的很多过程

os模块

os也就是操作系统，这里使用该库是为了访问环境变量获取api key，更安全

api_key = os.environ.get("TAVILY_API_KEY")

具体实现

import os
from tavily import TavilyClient

def get_attraction(city: str,weather: str) -> str:
    api_key = os.environ.get("TAVILY_API_KEY")
    if not api_key:
        return "错误:未配置TAVILY_API_KEY环境变量。"
    
    tavily = TavilyClient(api_key=api_key)

    query = f"'{city}'在'{weather}'天气下最值得去的旅游景点推荐"

    response = tavily.search(query=query, search_depth="basic", include_answer=True)
    print(response)
get_attraction('北京','Thundery outbreaks in nearby, 17°C')

直接输出的话格式会很乱，资源也很杂，所以可以用response["answer"]做一个简单的总结

输出结果：

1	Under thundery weather, visit the Temple of Heaven and the Forbidden City for cultural experiences. Beijing's best season for travel is autumn. For more details, check local weather forecasts.

完整实现

import os
from tavily import TavilyClient

def get_attraction(city: str,weather: str) -> str:
    api_key = os.environ.get("TAVILY_API_KEY")
    if not api_key:
        return "错误:未配置TAVILY_API_KEY环境变量。"
    
    tavily = TavilyClient(api_key=api_key)

    query = f"'{city}'在'{weather}'天气下最值得去的旅游景点推荐"

    try:
        response = tavily.search(query=query, search_depth="basic", include_answer=True)
    
        if response.get("answer"):
            return response["answer"]

        formatted_results = []
        for result in response.get("results",[]): 
            formatted_results.append(f"-{result['title']}:{result['content']}")

        if not formatted_results:
            return "抱歉，没有找到相关的旅游景点推荐。"
        return "根据搜索，为您找到以下信息:\n" + "\n".join(formatted_results)
    except Exception as e:
        return f"错误:执行Tavily搜索时出现问题 - {e}"

print(get_attraction('成都','晴天, 23°C'))

1 2	for result in response.get("results",[]): formatted_results.append(f"-{result['title']}:{result['content']}")

这一段是个简单的数据筛选

result是字典类型，results是列表类型

筛选出每一条result里的title字段，content字段

response.get("results",[]): 从 response 里拿名字叫 “results” 的内容

如果拿不到，就给我一个空列表 []

把上面两个工具注册

available_tools = {
    "get_weather": get_weather,
    "get_attraction": get_attraction,
}

通用的客户端 `OpenAICompatibleClient`接入大语言模型

当前，许多 LLM 服务提供商（包括 OpenAI、Azure、以及众多开源模型服务框架如 Ollama、vLLM 等）都遵循了与 OpenAI API 相似的接口规范。这种标准化为开发者带来了极大的便利。智能体的自主决策能力来源于 LLM。我们将实现一个通用的客户端 OpenAICompatibleClient，它可以连接到任何兼容 OpenAI 接口规范的 LLM 服务。

这里定义这个OpenAICompatibleClient 类后面调用大语言模型就可以直接用

llm_client = OpenAICompatibleClient(
    model="gpt-4o",
    api_key="sk-xxxxxx",
    base_url="https://api.openai.com/v1"
)

# 系统提示词（设定角色）
system_prompt = "你是一个专业旅游推荐官，回答简洁、中文、准确。"

# 用户问题（你要问什么）
user_prompt = "北京雷阵雨17度，推荐适合的景点"

# 让AI回答
answer = llm_client.generate(user_prompt, system_prompt)

# 打印结果
print("最终推荐：", answer)

首先写个构造函数，给工具”填参数、初始化”

class OpenAICompatibleClient:
    def __init__(self, model: str, api_key: str, base_url: str):
        self.model = model
        self.client = OpenAI(api_key=api_key, base_url=base_url)

python中写构造函数是def __init__()（注意是双下划线，前后各两个）

def __init__(self, model: str, api_key: str, base_url: str)

这里的self参数是之当前这个对象自己

这个构造函数负责接收外部数据然后传入当前类中

给客户端添加真正干活的功能，定义提示词，模型，回答

def generate(self, prompt: str, system_prompt: str) -> str:
        """调用LLM API来生成回应。"""
        print("正在调用大语言模型...")
        try:
            messages = [
                {'role': 'system', 'content': system_prompt},
                {'role': 'user', 'content': prompt}
            ]
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                stream=False
            )
            answer = response.choices[0].message.content
            print("大语言模型响应成功。")
            return answer
        except Exception as e:
            print(f"调用LLM API时发生错误: {e}")
            return "错误:调用语言模型服务时出错。"

首先messages列表，将提示词定义为系统提示词于用户提示词

“你是一个算法工程师”–>系统提示词

“我需要你写一个冒泡排序算法”–>用户提示词

response = self.client.chat.completions.create这里是真正调用ai接口的地方

self.client是我们在前面构造函数创建好的 OpenAI 客户端

.chat.completions.create是官方规定的聊天对话接口

然后就是指定用哪个模型，提示词是什么，输出是否为流式输出

完整实现

from openai import OpenAI

class OpenAICompatibleClient:
    """
    一个用于调用任何兼容OpenAI接口的LLM服务的客户端。
    """
    def __init__(self, model: str, api_key: str, base_url: str):
        self.model = model
        self.client = OpenAI(api_key=api_key, base_url=base_url)

    def generate(self, prompt: str, system_prompt: str) -> str:
        """调用LLM API来生成回应。"""
        print("正在调用大语言模型...")
        try:
            messages = [
                {'role': 'system', 'content': system_prompt},
                {'role': 'user', 'content': prompt}
            ]
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                stream=False
            )
            answer = response.choices[0].message.content
            print("大语言模型响应成功。")
            return answer
        except Exception as e:
            print(f"调用LLM API时发生错误: {e}")
            return "错误:调用语言模型服务时出错。"

整合整套路线，实现LLM循环

import re

# --- 1. 配置LLM客户端 ---
# 请根据您使用的服务，将这里替换成对应的凭证和地址
API_KEY = "YOUR_API_KEY"
BASE_URL = "YOUR_BASE_URL"
MODEL_ID = "YOUR_MODEL_ID"
TAVILY_API_KEY="YOUR_Tavily_KEY"
os.environ['TAVILY_API_KEY'] = "YOUR_TAVILY_API_KEY"

llm = OpenAICompatibleClient(
    model=MODEL_ID,
    api_key=API_KEY,
    base_url=BASE_URL
)

# --- 2. 初始化 ---
user_prompt = "你好，请帮我查询一下今天北京的天气，然后根据天气推荐一个合适的旅游景点。"
prompt_history = [f"用户请求: {user_prompt}"]

print(f"用户输入: {user_prompt}\n" + "="*40)

# --- 3. 运行主循环 ---
for i in range(5): # 设置最大循环次数
    print(f"--- 循环 {i+1} ---\n")
    
    # 3.1. 构建Prompt
    full_prompt = "\n".join(prompt_history)
    
    # 3.2. 调用LLM进行思考
    llm_output = llm.generate(full_prompt, system_prompt=AGENT_SYSTEM_PROMPT)
    # 模型可能会输出多余的Thought-Action，需要截断，每一轮之输出一对Thought-Action
    match = re.search(r'(Thought:.*?Action:.*?)(?=\n\s*(?:Thought:|Action:|Observation:)|\Z)', llm_output, re.DOTALL)
    if match:
        truncated = match.group(1).strip()
        if truncated != llm_output.strip():
            llm_output = truncated
            print("已截断多余的 Thought-Action 对")
    print(f"模型输出:\n{llm_output}\n")
    prompt_history.append(llm_output)
    
    # 3.3. 解析并执行行动
    action_match = re.search(r"Action: (.*)", llm_output, re.DOTALL)
    if not action_match:
        observation = "错误: 未能解析到 Action 字段。请确保你的回复严格遵循 'Thought: ... Action: ...' 的格式。"
        observation_str = f"Observation: {observation}"
        print(f"{observation_str}\n" + "="*40)
        prompt_history.append(observation_str)
        continue
    action_str = action_match.group(1).strip()

    if action_str.startswith("Finish"):
        final_answer = re.match(r"Finish\[(.*)\]", action_str).group(1)
        print(f"任务完成，最终答案: {final_answer}")
        break
    
    tool_name = re.search(r"(\w+)\(", action_str).group(1)
    args_str = re.search(r"\((.*)\)", action_str).group(1)
    kwargs = dict(re.findall(r'(\w+)="([^"]*)"', args_str))

    if tool_name in available_tools:
        observation = available_tools[tool_name](**kwargs)
    else:
        observation = f"错误:未定义的工具 '{tool_name}'"

    # 3.4. 记录观察结果
    observation_str = f"Observation: {observation}"
    print(f"{observation_str}\n" + "="*40)
    prompt_history.append(observation_str)

这里我们的提示词规定了模型的输出

```
AGENT_SYSTEM_PROMPT = """
你是一个智能旅行助手。你的任务是分析用户的请求，并使用可用工具一步步地解决问题。

# 可用工具:
- `get_weather(city: str)`: 查询指定城市的实时天气。
- `get_attraction(city: str, weather: str)`: 根据城市和天气搜索推荐的旅游景点。

# 输出格式要求:
你的每次回复必须严格遵循以下格式，包含一对Thought和Action：

Thought: [你的思考过程和下一步计划]
Action: [你要执行的具体行动]

Action的格式必须是以下之一：
1. 调用工具：function_name(arg_name="arg_value")
2. 结束任务：Finish[最终答案]

# 重要提示:
- 每次只输出一对Thought-Action
- Action必须在同一行，不要换行
- 当收集到足够信息可以回答用户问题时，必须使用 Action: Finish[最终答案] 格式结束

请开始吧！
"""
```

模型首先会接收用户需求，思考，然后执行，获得结果追加提示词，思考，执行，最终收集到足够信息之后Finish输出最终答案

用户输入: 你好，请帮我查询一下今天北京的天气，然后根据天气推荐一个合适的旅游景点。
==================================================
--- 循环 1 ---
正在调用大语言模型...
模型输出:
 Thought: 用户想要查询北京今天的天气，并根据天气推荐景点。首先需要获取北京的天气信息。
Action: get_weather(city="北京") 

Observation: 北京当前天气:Thundery outbreaks in nearby，气温25摄氏度
==================================================
--- 循环 2 ---
正在调用大语言模型...
模型输出:
 Thought: 已获取北京天气为"Thundery outbreaks in nearby"，需要基于此天气推荐景点。
Action: get_attraction(city="北京", weather="Thundery outbreaks in nearby") 

Observation: Under thundery weather, Beijing's Forbidden City and Tiananmen Square are best for indoor visits. Avoid outdoor attractions like the Great Wall.
==================================================
--- 循环 3 ---
正在调用大语言模型...
模型输出:
 Thought: 已经获取到基于当前天气的景点推荐，可以整合信息回答用户。
Action: Finish[今日北京天气为雷暴天气，气温25摄氏度，建议室内游览。推荐景点：故宫博物院、天安门广场。] 

✅ 任务完成！最终答案： 今日北京天气为雷暴天气，气温25摄氏度，建议室内游览。推荐景点：故宫博物院、天安门广场。

为什么要设置循环呢？因为任务太复杂AI不可能一步做完，就以这个推荐旅游景点为例，它就可以概括为三步：查天气、查景点、整理答案，让ai一步一步思考，一步一步执行，也设置了边界，只能循环5次，5次不行就中断并返回报错信息，调用工具必须去我们注册的工具里找，没有找到就不调用，并把每一次循环的输出追加到提示词中，最终整理好答案后输出结果

完整实现

travel_agent.py

# travel_agent.py —— 智能旅行助手（完整可运行版）
import requests
import re
import os
from openai import OpenAI
from tavily import TavilyClient

# ====================== 1. 系统提示词（智能体说明书）======================
AGENT_SYSTEM_PROMPT = """
你是一个智能旅行助手。你的任务是分析用户的请求，并使用可用工具一步步地解决问题。

# 可用工具:
- `get_weather(city: str)`: 查询指定城市的实时天气。
- `get_attraction(city: str, weather: str)`: 根据城市和天气搜索推荐的旅游景点。

# 输出格式要求:
你的每次回复必须严格遵循以下格式，包含一对Thought和Action：

Thought: [你的思考过程和下一步计划]
Action: [你要执行的具体行动]

Action的格式必须是以下之一：
1. 调用工具：function_name(arg_name="arg_value")
2. 结束任务：Finish[最终答案]

# 重要提示:
- 每次只输出一对Thought-Action
- Action必须在同一行，不要换行
- 当收集到足够信息可以回答用户问题时，必须使用 Action: Finish[最终答案] 格式结束

请开始吧！
"""

# ====================== 2. 工具1：查询天气 ======================
def get_weather(city: str) -> str:
    url = f"https://wttr.in/{city}?format=j1"
    try:
        response = requests.get(url)
        response.raise_for_status()
        data = response.json()
        current_condition = data['current_condition'][0]
        weather_desc = current_condition['weatherDesc'][0]['value']
        temp_c = current_condition['temp_C']
        return f"{city}当前天气:{weather_desc}，气温{temp_c}摄氏度"
    except Exception as e:
        return f"天气查询错误：{str(e)}"

# ====================== 3. 工具2：景点推荐 ======================
def get_attraction(city: str, weather: str) -> str:
    api_key = os.environ.get("TAVILY_API_KEY")
    if not api_key:
        return "错误：未配置TAVILY_API_KEY"
    tavily = TavilyClient(api_key=api_key)
    query = f"'{city}' 在'{weather}'天气下最值得去的旅游景点推荐及理由"
    try:
        response = tavily.search(query=query, search_depth="basic", include_answer=True)
        return response.get("answer", "未找到景点信息")
    except Exception as e:
        return f"景点搜索错误：{str(e)}"

# 工具注册
available_tools = {
    "get_weather": get_weather,
    "get_attraction": get_attraction,
}

# ====================== 4. LLM客户端 ======================
class OpenAICompatibleClient:
    def __init__(self, model: str, api_key: str, base_url: str):
        self.model = model
        self.client = OpenAI(api_key=api_key, base_url=base_url)

    def generate(self, prompt: str, system_prompt: str) -> str:
        print("正在调用大语言模型...")
        try:
            messages = [
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": prompt}
            ]
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                stream=False
            )
            return response.choices[0].message.content
        except Exception as e:
            return f"LLM调用错误：{str(e)}"

# ====================== 5. 配置区（你只需要改这里！）======================
API_KEY = "xxx"
BASE_URL = "https://api.deepseek.com"
MODEL_ID = "xxx"
TAVILY_API_KEY = "xxx"

os.environ['TAVILY_API_KEY'] = TAVILY_API_KEY

# 初始化LLM
llm = OpenAICompatibleClient(
    model=MODEL_ID,
    api_key=API_KEY,
    base_url=BASE_URL
)

# ====================== 6. 主循环：Thought-Action-Observation ======================
if __name__ == "__main__":
    user_prompt = "你好，请帮我查询一下今天北京的天气，然后根据天气推荐一个合适的旅游景点。"
    prompt_history = [f"用户请求: {user_prompt}"]

    print("用户输入:", user_prompt)
    print("=" * 50)

    # 最多循环5次
    for i in range(5):
        print(f"--- 循环 {i+1} ---")
        full_prompt = "\n".join(prompt_history)
        
        # LLM思考
        llm_output = llm.generate(full_prompt, AGENT_SYSTEM_PROMPT)
        print("模型输出:\n", llm_output, "\n")
        prompt_history.append(llm_output)

        # 解析Action
        action_match = re.search(r"Action: (.*)", llm_output)
        if not action_match:
            obs = "错误：无法解析Action"
            prompt_history.append(f"Observation: {obs}")
            continue
        action = action_match.group(1).strip()

        # 结束任务
        if action.startswith("Finish"):
            final = re.findall(r"Finish\[(.*?)\]", action)[0]
            print("✅ 任务完成！最终答案：", final)
            break

        # 执行工具
        tool_name = re.findall(r"(\w+)\(", action)[0]
        args = dict(re.findall(r'(\w+)="([^"]+)"', action))
        observation = available_tools[tool_name](**args)
        
        print("Observation:", observation)
        prompt_history.append(f"Observation: {observation}")
        print("=" * 50)

实现一个基于 API 调用的、最简单、最标准的 ReAct 智能体

get_weather()

requests库

json数据提取

异常处理

get_attraction

为什么用Tavily

os模块

具体实现

通用的客户端 OpenAICompatibleClient接入大语言模型

整合整套路线，实现LLM循环

通用的客户端 `OpenAICompatibleClient`接入大语言模型