一、前言

最近在学习 AI Agent，课程里讲到了一个非常经典的智能体运行范式：

Thought -> Action -> Observation

也就是：

思考 -> 行动 -> 观察结果 -> 再次思考

为了加深理解，基于hello-agent手动实现了一个最小版本的智能旅行助手。它可以根据用户输入，自动完成以下流程：

用户输入旅游需求 -> 大模型判断需要先查询天气 -> 调用天气查询工具 -> 根据天气调用景点推荐工具 -> 最后总结出旅游建议

这个案例没有直接使用 LangChain，而是手动实现 Agent 的核心循环，更适合初学者理解智能体底层到底是怎么工作的。

二、项目效果

运行程序后输入：

我想要去宿松旅游

程序会自动执行：

D:\anaconda\envs\helloagent\python.exe D:\hello-agent\main.py 
请输入你的旅行需求：我想要去宿松旅游
用户输入: 我想要去宿松旅游
========================================
--- 循环 1 ---

正在调用大语言模型...
大语言模型响应成功。
模型输出:
Thought: 用户想要去宿松旅游，我需要先查询宿松的天气情况，以便后续推荐景点。

Action: get_weather(city="宿松")

Observation: 错误:查询天气时遇到网络问题 - HTTPSConnectionPool(host='wttr.in', port=443): Max retries exceeded with url: /%E5%AE%BF%E6%9D%BE?format=j1 (Caused by SSLError(SSLEOFError(8, '[SSL: UNEXPECTED_EOF_WHILE_READING] EOF occurred in violation of protocol (_ssl.c:1017)')))
========================================
--- 循环 2 ---

正在调用大语言模型...
大语言模型响应成功。
已截断多余的 Thought-Action 对
模型输出:
Thought: 查询宿松天气时遇到了网络问题，可能是由于SSL协议问题或网络连接不稳定。我可以尝试再次查询，或者考虑使用其他方式获取信息。不过根据规则，我应该继续尝试查询天气。让我再试一次。

Action: get_weather(city="宿松")

Observation: 宿松当前天气:Moderate rain，气温20摄氏度
========================================
--- 循环 3 ---

正在调用大语言模型...
大语言模型响应成功。
已截断多余的 Thought-Action 对
模型输出:
Thought: 已经获取到宿松的天气情况（Moderate rain），现在可以根据城市和天气来搜索推荐的旅游景点。

Action: get_attraction(city="宿松", weather="Moderate rain")

Observation: 在中雨天气下，白崖寨是宿松最值得去的景点，因其历史悠久和独特的自然景观而闻名。天柱山地质公园也是不错的选择，适合户外活动和博物馆参观。
========================================
--- 循环 4 ---

正在调用大语言模型...
大语言模型响应成功。
模型输出:
Thought: 我已经获取到宿松在中雨天气下的推荐景点：白崖寨和天柱山地质公园，信息足够回答用户的问题。

Action: Finish[宿松目前天气为小雨（Moderate rain），气温20摄氏度。在中雨天气下，推荐您游览以下景点：\n1. **白崖寨** — 历史悠久，自然景观独特，是宿松最值得去的景点；\n2. **天柱山地质公园** — 适合户外活动和博物馆参观。\n请注意携带雨具，祝您旅途愉快！]

任务完成，最终答案: 宿松目前天气为小雨（Moderate rain），气温20摄氏度。在中雨天气下，推荐您游览以下景点：\n1. **白崖寨** — 历史悠久，自然景观独特，是宿松最值得去的景点；\n2. **天柱山地质公园** — 适合户外活动和博物馆参观。\n请注意携带雨具，祝您旅途愉快！

进程已结束，退出代码为 0

这就实现了一个简单但完整的 Agent。

三、环境准备

我使用的是 conda 环境：

conda create -n helloagent python=3.10 conda activate helloagent

安装依赖：

pip install requests tavily-python openai

其中：

requests：用于调用天气 API tavily-python：用于联网搜索景点推荐 openai：用于调用兼容 OpenAI 格式的大模型接口

四、项目结构

一开始可以把所有代码都写在 main.py 里，但为了更清晰，我将项目拆成了多个文件：

hello-agent/
├── main.py
├── config.py
├── prompts.py
├── tools.py
└── llm_client.py

每个文件的职责如下：

config.py：保存 API 配置

prompts.py：保存系统提示词

tools.py：保存工具函数

llm_client.py：封装大模型调用

main.py：运行 Agent 主循环

五、配置文件 config.py

API_KEY = "你的大模型API_KEY"

BASE_URL = "你的大模型BASE_URL"

MODEL_ID = "你的模型名称"

TAVILY_API_KEY = "你的Tavily API_KEY"

例如，我使用的是deepseek：

API_KEY = "xxxxxxxx"
BASE_URL = "https://api.deepseek.com"
MODEL_ID = "deepseek-chat"

TAVILY_API_KEY = "xxxxxxxxxxx"

注意：真实项目中不要把 config.py 上传到 GitHub。

六、提示词 prompts.py

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[最终答案] 格式结束
- 最终回答请使用简体中文

请开始吧！
"""

这个提示词非常关键，它告诉大模型：

你是谁 你能用哪些工具 你应该按什么格式输出 什么时候结束任务

七、工具函数 tools.py

import os
import requests
from tavily import TavilyClient


def get_weather(city: str) -> str:
    """
    通过调用 wttr.in API 查询真实的天气信息。
    """
    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 requests.exceptions.RequestException as e:
        return f"错误:查询天气时遇到网络问题 - {e}"
    except (KeyError, IndexError) as e:
        return f"错误:解析天气数据失败，可能是城市名称无效 - {e}"


def get_attraction(city: str, weather: str) -> str:
    """
    根据城市和天气，使用 Tavily Search API 搜索并返回景点推荐。
    """
    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}"


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

这里定义了两个工具：

get_weather：查询天气 get_attraction：根据天气搜索景点

最后用字典统一管理工具，方便主程序根据大模型输出的 Action 动态调用。

八、大模型客户端 llm_client.py

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 "错误:调用语言模型服务时出错。"

这里封装了大模型调用逻辑。

只要服务商兼容 OpenAI API 格式，就可以使用这种方式调用。

九、主程序 main.py

import os
import re

from config import API_KEY, BASE_URL, MODEL_ID, TAVILY_API_KEY
from llm_client import OpenAICompatibleClient
from prompts import AGENT_SYSTEM_PROMPT
from tools import available_tools


os.environ["TAVILY_API_KEY"] = TAVILY_API_KEY


def run_agent(user_prompt: str, max_rounds: int = 5):
    llm = OpenAICompatibleClient(
        model=MODEL_ID,
        api_key=API_KEY,
        base_url=BASE_URL,
    )

    prompt_history = [f"用户请求: {user_prompt}"]

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

    for i in range(max_rounds):
        print(f"--- 循环 {i + 1} ---\n")

        full_prompt = "\n".join(prompt_history)

        llm_output = llm.generate(
            full_prompt,
            system_prompt=AGENT_SYSTEM_PROMPT,
        )

        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)

        action_match = re.search(r"Action: (.*)", llm_output, re.DOTALL)
        if not action_match:
            observation = "错误: 未能解析到 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"):
            finish_match = re.match(r"Finish\[(.*)\]", action_str, re.DOTALL)
            if finish_match:
                final_answer = finish_match.group(1)
                print(f"任务完成，最终答案: {final_answer}")
                return final_answer

            print("错误: Finish 格式不正确。")
            return None

        tool_name_match = re.search(r"(\w+)\(", action_str)
        args_match = re.search(r"\((.*)\)", action_str)

        if not tool_name_match or not args_match:
            observation = f"错误: 无法解析工具调用: {action_str}"
            observation_str = f"Observation: {observation}"
            print(f"{observation_str}\n" + "=" * 40)
            prompt_history.append(observation_str)
            continue

        tool_name = tool_name_match.group(1)
        args_str = args_match.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}'"

        observation_str = f"Observation: {observation}"
        print(f"{observation_str}\n" + "=" * 40)
        prompt_history.append(observation_str)

    print("达到最大循环次数，任务未完成。")
    return None


if __name__ == "__main__":
    user_prompt = input("请输入你的旅行需求：").strip()

    if not user_prompt:
        user_prompt = "你好，请帮我查询一下今天合肥的天气，然后根据天气推荐一个合适的旅游景点。"

    run_agent(user_prompt)

十、运行程序

在 PyCharm 中直接运行 main.py。

输入：

我想要去宿松旅游

程序输出类似：

用户输入: 我想要去宿松旅游 ======================================== --- 循环 1 --- 正在调用大语言模型... 大语言模型响应成功。 模型输出: Thought: 用户想要去宿松旅游，我需要先查询宿松的天气情况。 Action: get_weather(city="宿松") Observation: 宿松当前天气:Moderate rain，气温20摄氏度 ======================================== --- 循环 2 --- 模型输出: Thought: 已经获取到宿松的天气情况，现在可以根据天气推荐景点。 Action: get_attraction(city="宿松", weather="Moderate rain") Observation: 在中雨天气下，白崖寨是宿松最值得去的景点。 ======================================== --- 循环 3 --- Action: Finish[宿松目前天气为中雨，气温20摄氏度。推荐您游览白崖寨，请注意携带雨具。]

十一、遇到的问题

1. 大模型接口 404

如果出现：

调用LLM API时发生错误: Error code: 404

一般是：

BASE_URL 填错了 MODEL_ID 填错了

解决方法是去服务商文档中查看正确的：

OpenAI compatible base_url 模型名称

2. Tavily API Key 无效

如果出现：

Invalid API key: Unauthorized: missing or invalid API key

说明 Tavily 的 Key 没有配置正确。

检查：

TAVILY_API_KEY = "你的Tavily API_KEY"

以及：

os.environ["TAVILY_API_KEY"] = TAVILY_API_KEY

3. 天气接口偶尔失败

有时 wttr.in 会出现网络或 SSL 错误。这个案例中，Agent 可以根据 Observation 再次尝试调用工具。

这也体现了 Agent 的特点：

工具失败 -> 观察失败原因 -> 决定下一步动作 -> 继续尝试或给出替代方案

十二、总结

这个案例虽然简单，但已经包含了 Agent 的核心思想：

1. 大模型负责思考和决策 2. 工具负责执行具体任务 3. Observation 将工具结果反馈给大模型 4. 多轮循环让 Agent 可以逐步完成复杂任务

整个流程可以概括为：

用户问题 -> LLM 输出 Thought 和 Action -> Python 解析 Action -> 调用对应工具 -> 得到 Observation -> 再交给 LLM -> 最终 Finish

通过这个案例，我对 AI Agent 的理解更清晰了。LangChain、LlamaIndex 等框架本质上也是在这个思想基础上做了更强大的封装。对于初学者来说，先手写一个最小 Agent，比一开始直接使用框架更容易理解底层原理。