前言

在人工智能的浪潮中,我们见证了从简单的聊天机器人Chatbot到能够自主思考、调用工具的智能Agent的进化。本文将结合DataWhale的hello-agents教程，带你深入理解Agent的核心原理,并以OpenAI SDK和经典的LLM应用开发框架——Langchain两种方式，演示如何从零开始构建一个智能旅行助手Agent。

---

一、什么是Agent?Agent与LLM的本质区别

1.1 传统LLM的局限性

传统的大语言模型(LLM)虽然强大,但存在明显的局限性:

• 知识截止:训练数据有时间边界,无法获取实时信息
• 无法行动:只能生成文本,无法与外部世界交互
• 幻觉问题:可能编造不存在的事实

1.2 Agent的突破

Agent通过引入**工具调用(Tool Calling)**能力,让LLM能够:

• 🌐 获取实时信息:通过API调用最新数据，尽可能避免幻觉。
• 🛠️ 执行具体操作:不只是说,而是真正去做，给发达的”大脑“加上手脚
• 🔄 多步推理:通过"思考-行动-观察"的循环解决复杂问题

核心理念:Agent = LLM + 工具(Tools) + 推理循环(Reasoning Loop)

---

二、Agent的工作原理:ReAct框架

2.1 ReAct范式解析

ReAct(Reasoning + Acting)是Agent设计的核心框架:

1. Thought(思考):分析当前状态,决定下一步
2. Action(行动):调用特定工具
3. Observation(观察):获取工具返回结果
4. 循环:回到步骤1,直到完成任务

2.2 系统提示词的设计艺术

一个精心设计的System Prompt至关重要:

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

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

# 行动格式:
你的回答必须严格遵循以下格式。首先是你的思考过程,然后是你要执行的具体行动,每次回复只输出一对Thought-Action:
Thought: [这里是你的思考过程和下一步计划]
Action: [这里是你要调用的工具,格式为 function_name(arg_name="arg_value")]

# 任务完成:
当你收集到足够的信息,能够回答用户的最终问题时,你必须在`Action:`字段后使用 `finish(answer="...")` 来输出最终答案。
"""

设计要点:

1. 明确的角色定位(System):告诉AI它是什么,要做什么
2. 工具清单(Tools):清晰列出可用工具及其功能
3. 输出格式约束(Output):严格的结构化输出要求，便于进一步处理(例如可将大模型返回的信息直接作为参数，交由其他函数模块处理)
4. 终止条件(Stop):明确何时结束循环

---

三、实战演练:从零构建旅行助手Agent

3.1 工具层设计:连接真实世界

工具1:天气查询API集成

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 Exception as e:
        return f"错误:查询天气时遇到网络问题 - {e}"

此处返回JSON，并且这种设计便于更换API。

工具2:智能景点推荐

def get_attraction(city: str, weather: str) -> str:
    """根据城市和天气,使用Tavily Search API搜索并返回优化后的景点推荐"""
    api_key = os.environ.get("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
        )

        # Tavily已经为我们总结了搜索结果
        if response.get("answer"):
            return response["answer"]

        # 如果没有总结,则格式化原始结果
        formatted_results = []
        for result in response.get("results", []):
            formatted_results.append(f"- {result['title']}: {result['content']}")

        return "根据搜索,为您找到以下信息:\n" + "\n".join(formatted_results)
    except Exception as e:
        return f"错误:执行Tavily搜索时出现问题 - {e}"

Tavily是专门适配于Agent的搜索引擎，相较于google，baidu等引擎，它更加友好，更加干净，便于配合Agent行为

3.2 推理引擎:LLM作为大脑

兼容多种LLM的客户端设计

class OpenAICompatibleClient:
    """一个用于调用任何兼容OpenAI接口的LLM服务的客户端"""
    def __init__(self, model: str="deepseek-chat", base_url: str="https://api.deepseek.com"):
        self.model = model
        self.client = ChatOpenAI(
            model=self.model,
            api_key=os.getenv("OPENAI_API_KEY"),
            base_url=base_url
        )

    def generate(self, prompt: str, system_prompt: str):
        """调用LLM API来生成回应"""
        try:
            full_prompt = f"{system_prompt}\n\n用户: {prompt}"
            response = self.client.invoke(input=full_prompt, stream=False)
            return response.content
        except Exception as e:
            return f"错误:调用语言模型服务时出错。"

此处支持任何OpenAI兼容的API(DeepSeek、通义千问等)，国产大模型的API更便于调用

3.3 核心循环:思考-行动-观察

# 主循环逻辑
for i in range(5):  # 设置最大循环数
    print(f"--- 循环 {i+1} ---")

    # 1. 构建prompt(包含历史对话)
    full_prompt = "\n".join(prompt_history)

    # 2. 调用大模型获取决策
    llm_output = llm.generate(full_prompt, system_prompt=AGENT_SYSTEM_PROMPT)

    # 3. 解析模型输出的Action
    action_match = re.search(r"Action: (.*)", llm_output, re.DOTALL)
    if not action_match:
        print("解析错误:模型输出中未找到 Action。")
        break

    action_str = action_match.group(1).strip()

    # 4. 检查是否完成任务
    if action_str.startswith("finish"):
        final_answer = re.search(r'finish\(answer="(.*)"\)', action_str).group(1)
        print(f"任务完成,最终答案: {final_answer}")
        break

    # 5. 解析工具调用参数
    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))

    # 6. 执行工具调用
    if tool_name in tools:
        observation = tools[tool_name](**kwargs)
    else:
        observation = f"错误:未定义的工具 '{tool_name}'"

    # 7. 记录观察结果
    observation_str = f"Observation: {observation}"
    prompt_history.append(observation_str)

运行示例:

用户输入: 请帮我查询今天纽约的天气,然后根据天气推荐几个合适的旅游景点

--- 循环 1 ---
Thought: 用户需要查询纽约的天气,并根据天气推荐旅游景点。我需要先调用 `get_weather` 工具获取纽约的天气信息。
Action: get_weather(city="纽约")
Observation: 纽约当前天气:晴,气温15摄氏度

--- 循环 2 ---
Thought: 已获取天气信息,现在需要根据晴朗的天气推荐景点。
Action: get_attraction(city="纽约", weather="晴")
Observation: 根据搜索,为您找到以下信息:
- 中央公园:晴天适合散步、野餐
- 自由女神像:户外参观的好时机
- 布鲁克林大桥:天气晴朗,适合步行观景

--- 循环 3 ---
Thought: 已收集足够信息,可以给出最终答案。
Action: finish(answer="纽约今天天气晴朗,气温15度,非常适合户外活动。推荐景点:1)中央公园-适合散步野餐;2)自由女神像-户外参观;3)布鲁克林大桥-步行观景绝佳选择。")
任务完成!

此处有几个设计亮点
关键设计决策:

1. 输出截断机制:

# 防止模型输出多个Thought-Action对
match = re.search(
    r'(Thought:.*?Action:.*?)(?=\n\s*(?:Thought:|Action:|Observation:)|\Z)',
    llm_output,
    re.DOTALL
)

2. 安全循环限制:设置最大循环次数防止无限循环
3. 错误容错:工具调用失败时继续执行,最终用常识兜底

---

四、实战优化:使用LangChain框架

手动实现Agent虽然有助于理解原理,但生产环境推荐使用成熟框架。让我们看看如何用LangChain实现相同功能。

4.1 工具定义升级

在国内，为了保证查询的有效性，可以选择国内的天气查询API来构建工具

from langchain_core.tools import StructuredTool

# 将函数包装为LangChain工具
# StructedTool.from_function可将一个函数直接转化为一个Langchain的工具类
attraction_tool = StructuredTool.from_function(
    func=get_attraction,
    name="get_attraction",
    description="根据天气和城市,查询数个推荐的景点"
)

# 使用@tool装饰器定义
@tool
#@tool
def weather_tool(city: str) -> str:
    """
    使用和风天气 API 查询指定城市的实时天气。
    
    需要环境变量:
    - QWEATHER_API_KEY: 和风天气 API Key
    
    Args:
        city: 城市名称，支持中文（如"北京"、"上海"）或城市ID
        
    Returns:
        天气信息字符串
    """
    # 1. 获取 API Key
    api_key = os.environ.get("QWEATHER_API_KEY")
    if not api_key:
        return "错误:未配置 QWEATHER_API_KEY 环境变量。请先在和风天气官网申请 API Key。"
    
    # 2. 城市名称转城市ID（和风天气需要城市ID）
    # 使用城市搜索 API 获取城市ID
    city_search_url = f""  # 此处的URL可在和风天气控制台中查询
    
    try:
        # 搜索城市
        search_response = requests.get(city_search_url, timeout=10)
        search_response.raise_for_status()
        search_data = search_response.json()
        
        # 检查是否找到城市
        if search_data.get('code') != '200' or not search_data.get('location'):
            return f"错误:未找到城市 '{city}'，请检查城市名称是否正确。"
        
        # 获取第一个匹配城市的ID
        city_id = search_data['location'][0]['id']
        city_name = search_data['location'][0]['name']
        
        # 3. 获取实时天气
        weather_url = f"" # 此处url也可以在和风天气控制台中获得
        weather_response = requests.get(weather_url, timeout=10)
        weather_response.raise_for_status()
        weather_data = weather_response.json()
        
        # 检查响应状态
        if weather_data.get('code') != '200':
            return f"错误:获取天气数据失败 - {weather_data.get('code')}"
        
        # 4. 解析天气数据（虽然用不到这么多，但是都可以先写上，以便于扩展）
        now = weather_data['now']
        temp = now['temp']          # 温度
        feels_like = now['feelsLike']  # 体感温度
        text = now['text']          # 天气现象
        wind_dir = now['windDir']   # 风向
        wind_scale = now['windScale']  # 风力等级
        humidity = now['humidity']  # 湿度
        obs_time = now['obsTime']   # 观测时间
        
        # 格式化返回结果
        result = f"""
🌍 {city_name}当前天气:
☁️ 天气: {text}
🌡️ 温度: {temp}°C (体感 {feels_like}°C)
💨 风向风力: {wind_dir} {wind_scale}级
💧 湿度: {humidity}%
🕐 更新时间: {obs_time}
        """.strip()
        
        return result
        
    except requests.exceptions.Timeout as e:
        return f"错误:请求超时，网络连接较慢，请稍后重试。- {e}"
    except requests.exceptions.RequestException as e:
        return f"错误:网络请求失败 - {e}"
    except (KeyError, IndexError) as e:
        return f"错误:解析天气数据失败 - {e}"
    except Exception as e:
        return f"错误:未知错误 - {e}"
agent_tools = [weather_tool, attraction_tool]

4.2 模型兼容性

from langchain_openai.chat_models import ChatOpenAI
from langchain_community.chat_models import ChatTongyi

# 支持多种模型
model = ChatTongyi(
    model="qwen-max",
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

# 或使用DeepSeek
# model = ChatOpenAI(
#     model="deepseek-chat",
#     api_key=os.getenv("OPENAI_API_KEY"),
#     base_url="https://api.deepseek.com"
# )

4.3 Agent创建与调用

from langchain.agents import create_agent

# 创建Agent
agent = create_agent(model=model, tools=agent_tools)

# 调用Agent
response = agent.invoke({
    "messages": [("user", "查询太原的天气并推荐景点")]
})

# 遍历消息流
for msg in response["messages"]:
    if msg.type == "human":
        print(f"用户: {msg.content}")
    elif msg.type == "ai":
        if hasattr(msg, 'tool_calls') and msg.tool_calls:
            print(f"AI调用工具: {msg.tool_calls[0]['name']}")
        else:
            print(f"AI回复: {msg.content}")
    elif msg.type == "tool":
        print(f"工具返回: {msg.content[:100]}...")

4.4 流式输出:更好的用户体验

# 流式调用Agent
inputs = {
    "messages": [("user", "查询运城的天气并推荐景点")]
}

for chunk in agent.stream(inputs, stream_mode="updates"):
    for node, values in chunk.items():
        messages = values["messages"]

        for msg in messages:
            if msg.type == "ai":
                if hasattr(msg, 'tool_calls') and msg.tool_calls:
                    for tool_call in msg.tool_calls:
                        print(f"🤖 AI思考: 调用工具 '{tool_call['name']}'")
                        print(f"   参数: {tool_call['args']}")
                elif msg.content:
                    print(f"🤖 AI回复: {msg.content}")

            elif msg.type == "tool":
                preview = str(msg.content)[:50] + "..."
                print(f"🔧 工具返回: {preview}")

输出效果:

🚀 开始运行 Agent...

🤖 AI思考: 调用工具 'weather_tool'
   参数: {'city': '运城'}
🔧 工具返回: 🌍 运城当前天气: ☁️ 天气: 晴 🌡️ 温度: 4°C (体感 -2°C)...

🤖 AI思考: 调用工具 'get_attraction'
   参数: {'city': '运城', 'weather': '晴'}
🔧 工具返回: 根据运城晴朗的天气,推荐以下景点...

==============================
🤖 AI回复: 运城今天天气晴朗,气温4度(体感-2度),建议注意保暖...

流式输出的价值:

• ⚡ 实时反馈:用户立即看到Agent的思考过程
• 🎭 透明性:展示决策链路,增强信任感
• 🎯 调试友好:清晰看到每一步的输入输出

---

五、深度思考:Agent设计的核心挑战与解决方案

5.1 工具调用的可靠性问题

挑战:Agent可能选择错误的工具或参数

解决方案:

1. 精确的工具描述:

# ❌ 模糊描述
def get_data(data):
    """获取数据"""

# ✅ 精确描述
def get_weather(city: str) -> str:
    """
    查询指定城市的实时天气信息。

    Args:
        city: 城市名称,支持中文(如"北京")或英文(如"Beijing")

    Returns:
        包含温度、天气状况、湿度等信息的字符串
    """

2. 参数验证:

def get_attraction(city: str, weather: str) -> str:
    # 参数校验
    if not city or len(city) < 2:
        return "错误:城市名称无效"

    # 设置默认天气
    if not weather:
        weather = get_weather(city)

    # 业务逻辑...

3. 工具输出的标准化:

# 统一错误格式
ERROR_PREFIX = "错误:"

def call_tool(tool_func, **kwargs):
    try:
        result = tool_func(**kwargs)
        if result.startswith(ERROR_PREFIX):
            # 记录错误日志
            log_error(result)
        return result
    except Exception as e:
        return f"{ERROR_PREFIX}{tool_func.__name__}异常 - {e}"

5.2 幻觉问题的缓解

挑战:Agent可能编造工具调用结果

多层防护策略:

1. 工具层验证:

def get_weather(city: str) -> str:
    response = requests.get(f"https://wttr.in/{city}?format=j1")

    # 验证响应
    if response.status_code != 200:
        return f"错误:API返回状态码 {response.status_code}"

    data = response.json()

    # 验证数据完整性
    if 'current_condition' not in data:
        return "错误:天气数据格式异常"

    # 数据提取...

2. Prompt工程强化:

SYSTEM_PROMPT = """
# 重要约束:
- 你只能使用列出的工具,不能编造工具
- 如果工具返回错误信息,必须如实反馈给用户
- 不得编造或猜测工具调用的结果
- 当且仅当通过工具获得真实数据后,才能给出最终答案
"""

3. 结果溯源:

def get_attraction(city: str, weather: str) -> str:
    response = tavily.search(query=query, include_answer=True)

    # 添加来源信息
    answer = response.get("answer", "")
    sources = [r['url'] for r in response.get('results', [])]

    return f"{answer}\n\n来源: {', '.join(sources)}"

5.3 性能优化策略

挑战:多次工具调用导致延迟

优化方案:

1. 并行调用:

# LangChain自动优化
agent = create_agent(
    model=model,
    tools=agent_tools,
    parallel_tool_calls=True  # 允许并行调用独立工具
)

2. 智能缓存:

from functools import lru_cache

@lru_cache(maxsize=100)
def get_weather(city: str) -> str:
    # 相同城市1小时内只调用一次API
    return _call_weather_api(city)

3. 流式处理:

# 快速展示部分结果
for chunk in agent.stream(inputs):
    # 先返回天气,再返回景点推荐
    if "weather" in chunk:
        yield chunk["weather"]
    if "attractions" in chunk:
        yield chunk["attractions"]

---

六、应用场景拓展:Agent的无限可能

6.1 企业级应用

1. 客服Agent:

   • 订单查询工具
   • 退款处理工具
   • FAQ知识库搜索

2. 数据分析Agent:

   • SQL查询工具
   • 数据可视化工具
   • 报告生成工具

3. 代码助手Agent:

   • 代码搜索工具
   • 单元测试生成工具
   • 文档生成工具

6.2 个人助手

1. 学习Agent:

   • 论文检索工具
   • 概念解释工具
   • 习题生成工具

2. 生活Agent:

   • 日程管理工具
   • 菜谱推荐工具
   • 健康管理工具

---

七、未来展望:Agent的发展方向

7.1 多模态Agent

当前Agent主要处理文本,未来将支持:

• 🖼️ 图像理解:视觉问答、图像描述
• 🎵 音频处理:语音交互、音乐生成
• 🎬 视频分析:视频问答、内容理解

7.2 自主学习Agent

• 📚 从反馈中学习:根据用户评价优化策略
• 🔄 动态工具发现:自动发现和注册新工具
• 🧠 知识积累:长期记忆与经验总结

7.3 协作Agent生态系统

• 👥 多Agent协作:专业分工,协同工作
• 🌐 Agent市场:工具和能力的交易市场
• 🔗 跨平台集成:无缝对接各类服务

---

八、总结与行动建议

核心要点回顾

1. Agent = LLM + 工具 + 推理循环
2. ReAct框架:思考→行动→观察的闭环
3. 三大支柱:
   • 精心设计的System Prompt
   • 可靠的工具层
   • 健壮的执行循环

学习路径建议

初级:

• ✅ 理解ReAct框架原理
• ✅ 手动实现简单Agent
• ✅ 掌握工具定义方法

中级:

• 🔧 使用LangChain等框架
• 🔧 处理错误和边界情况
• 🔧 优化Prompt和工具描述

高级:

• 🚀 设计多Agent系统
• 🚀 实现自定义工具链
• 🚀 构建生产级Agent应用

实战项目建议

1. 个人知识库Agent:

   • 工具:文档搜索、向量检索、总结生成
   • 场景:快速找到并总结个人笔记

2. 股票分析Agent:

   • 工具:实时行情、财报分析、新闻检索
   • 场景:智能投资建议

3. 代码审查Agent:

   • 工具:代码搜索、静态分析、最佳实践库
   • 场景:自动化代码审查

---

九、资源与延伸阅读

技术文档:

• LangChain官方文档
• OpenAI Function Calling
• ReAct论文原文

---

结语

从最初的简单聊天机器人到如今能够自主思考、调用工具的智能Agent,我们正在见证AI技术的质变。Agent不仅仅是技术的进步,更是AI从"对话"到"行动"的关键跨越。

掌握Agent开发,意味着你能够让AI:

• 🔍 获取真实世界的信息
• 🛠️ 执行具体的任务操作
• 🧠 进行复杂的多步推理

这正是构建下一代AI应用的核心能力。

现在就开始你的Agent之旅吧! 🚀

---

💡 本文所有代码均来自实战项目,已验证可运行。建议读者跟随代码实践,加深理解。

📧 如有问题或建议,欢迎讨论交流!

---