AI Agent 实例开发实战：从零构建你的第一个智能代理

1次阅读

没有评论

共计 3251 个字符，预计需要花费 9 分钟才能阅读完成。

传统脚本在处理固定流程任务时表现优异，但面对需要动态决策的场景就显得力不从心。比如天气预报查询，如果只是简单调用 API 返回数据，脚本完全可以胜任。但如果我们想要实现这样的功能：

根据用户输入自动判断是否需要查询天气（比如用户说 ” 明天需要带伞吗 ”）
能够处理模糊地点（如 ” 我家附近 ” 需要结合用户位置信息）
支持多轮对话（追问某地未来一周天气）

这时候传统脚本就需要写大量条件判断，而 AI Agent 通过结合 LLM 的推理能力和预定义工具集，可以更优雅地解决这类问题。

直接调用大语言模型 API 开发 Agent 会遇到：

提示工程 (Prompt Engineering) 复杂：需要精心设计 system message 和 few-shot 示例
输出解析困难：LLM 返回的非结构化文本需要复杂正则匹配
状态管理缺失：多轮对话需要自行维护上下文

LangChain 框架提供了以下关键抽象：

工具(Tools)：将 API 封装成可描述、可调用的标准化接口
代理(Agents)：内置多种决策策略(ReAct, Self-ask 等)
记忆(Memory)：自动管理对话历史

下面我们通过天气预报 Agent 实例来具体演示。

weather_agent/
├── tools.py        # 工具定义
├── agent.py        # 代理核心逻辑
└── config.py       # 配置管理

在 tools.py 中定义天气查询工具：

import requests
from typing import Optional
from pydantic import BaseModel

class WeatherInput(BaseModel):
    location: str
    date: Optional[str] = None

class WeatherTool:
    name = "get_weather"
    description = "查询指定地点和日期的天气情况"
    args_schema = WeatherInput

    @classmethod
    def run(cls, location: str, date: str = None):
        """调用天气 API 并格式化返回结果"""
        # 实际项目应该从配置读取 API KEY
        api_key = "your_api_key"
        base_url = "https://api.weatherapi.com/v1/forecast.json"

        params = {
            "key": api_key,
            "q": location,
            "days": 1 if not date else 7,
            "aqi": "no",
            "alerts": "no"
        }

        try:
            response = requests.get(base_url, params=params)
            response.raise_for_status()
            data = response.json()

            # 简化返回结构
            if date:
                forecast = next((day for day in data["forecast"]["forecastday"] 
                     if day["date"] == date),
                    None
                )
                return f"{location}{date}天气: {forecast['day']['condition']['text']}"
            else:
                current = data["current"]
                return f"{location}当前天气: {current['condition']['text']}, 温度{current['temp_c']}℃"

        except Exception as e:
            return f"天气查询失败: {str(e)}"

在 agent.py 中初始化代理：

from langchain.agents import AgentType, initialize_agent
from langchain.chat_models import ChatOpenAI
from tools import WeatherTool

class WeatherAgent:
    def __init__(self):
        self.llm = ChatOpenAI(
            temperature=0,
            model_name="gpt-3.5-turbo"
        )

        self.tools = [WeatherTool]

        self.agent = initialize_agent(
            tools=self.tools,
            llm=self.llm,
            agent=AgentType.CHAT_ZERO_SHOT_REACT_DESCRIPTION,
            verbose=True,
            handle_parsing_errors=True
        )

    def query(self, input_text: str) -> str:
        """执行用户查询"""
        try:
            return self.agent.run(input_text)
        except Exception as e:
            return f"Agent 执行出错: {str(e)}"

在 config.py 中处理敏感信息：

import os
from dotenv import load_dotenv

load_dotenv()

class Config:
    OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
    WEATHER_API_KEY = os.getenv("WEATHER_API_KEY")

    @classmethod
    def validate(cls):
        if not cls.OPENAI_API_KEY:
            raise ValueError("Missing OPENAI_API_KEY in .env")
        if not cls.WEATHER_API_KEY:
            raise ValueError("Missing WEATHER_API_KEY in .env")

为 WeatherTool 添加 @rate_limited 装饰器
使用 Tenacity 库实现自动重试

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type
)
import requests

@retry(stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10),
    retry=retry_if_exception_type(requests.exceptions.RequestException)
)
def safe_api_call(url, params):
    response = requests.get(url, params=params)
    response.raise_for_status()
    return response