手把手教你搭建一个智能代理：从基础代码到优化实践

一、引言

在人工智能领域，智能代理（Agent）是一个能够自主决策、调用工具并与用户交互的程序实体。它可以根据用户的查询，决定是否需要调用工具（如搜索实时信息、查询天气、操作数据库等），并将结果整理成自然语言回答。本文将通过一个具体的代码案例，详细介绍如何搭建一个智能代理，并对其进行优化，适合刚接触AI开发的初学者参考。

二、原始代码：实现基础功能的流水账Agent

我们先来看一个基础版本的智能代理代码，它实现了以下核心功能：

• 工具调用：支持搜索实时新闻、查询天气、插入用户信息到数据库。
• 状态管理：使用状态图（StateGraph）管理对话流程，决定何时调用工具、何时返回回答。
• LLM集成：基于LangChain和OpenAI/GPT模型处理自然语言请求。

关键代码解析

1. 工具定义：

   • 使用@tool装饰器定义工具函数，每个工具对应一个Pydantic模型（如SearchQuery、WeatherLoc），用于校验参数格式。
   • 示例：查询天气的工具函数get_weather，根据输入的城市返回预设天气信息。

     @tool(args_schema = WeatherLoc)
     def get_weather(location):
     if location.lower() in ["beijing"]:
        return "北京的温度是16度，天气晴朗。"
     # 其他城市逻辑...

2. 状态图构建：

   • 使用StateGraph定义对话流程：
     • 入口节点chat_with_model：调用LLM生成结构化输出（判断是否需要工具）。
     • 条件分支：如果需要工具，进入execute_function节点；否则进入final_answer节点。

       graph.add_conditional_edges(
       "chat_with_model",
       generate_branch,  # 函数判断是否需要工具
       {True: "execute_function", False: "final_answer"}
       )

3. 数据库操作：

   • 使用SQLAlchemy定义用户表模型User，并实现insert_db工具函数，用于向数据库插入用户信息。

三、优化升级：让Agent更健壮、易维护

原始代码虽然能工作，但在代码质量、健壮性和可维护性方面有提升空间。以下是优化后的版本及改进点：

优化点1：代码结构与可读性

• 分组导入与注释：将导入语句按模块分组（如标准库、第三方库、自定义模块），并添加注释说明功能。
• 命名规范：使用更清晰的变量名（如NEWS_API_KEY代替NEWS_API），工具函数参数使用Pydantic模型传递（如insert_db(user_info: UserInfo)）。
• 移除冗余：删除未使用的模块（如END、operator的部分引用），简化代码结构。

优化点2：错误处理与健壮性

• 网络请求增强：

  • 添加超时设置（timeout=10），避免请求挂起。
  • 使用response.raise_for_status()捕获HTTP错误，并返回友好的错误信息。

    try:
    response = requests.get(url, params=params, timeout=10)
    response.raise_for_status()  # 抛出HTTP错误
    # 处理响应...
    except requests.exceptions.RequestException as e:
    return json.dumps({"error": f"网络请求错误: {str(e)}"}, ensure_ascii=False)

• 数据库唯一性校验：

  • 在插入用户信息前，检查邮箱是否已存在，避免重复数据。

    existing_user = session.query(User).filter_by(email=user_info.email).first()
    if existing_user:
    return {"messages": [f"插入失败：邮箱地址 {user_info.email} 已存在"]}

优化点3：性能与配置管理

• 数据库连接池：

  • 通过pool_size、max_overflow等参数配置连接池，提升数据库操作性能。

    engine = create_engine(DATABASE_URI, pool_size=5, max_overflow=10, pool_timeout=30)

• LLM重试机制：

  • 在ChatOpenAI中添加max_retries=3，自动重试失败的API请求。

    llm = ChatOpenAI(
    model=MODEL_NAME,
    temperature=0,
    base_url=BASE_URL,
    api_key=OPENROUTER_API_KEY,
    max_retries=3  # 自动重试3次
    )

优化点4：类型注解与代码规范

• 完整类型声明：为函数参数和返回值添加类型注解（如def fetch_real_time_info(query: str) -> str:），提升代码可读性和IDE支持。
• Pydantic模型优化：在UserInfo模型中添加nullable=False等约束，明确字段必填性。

优化点5：可测试性与示例

• 添加测试函数：编写test_agent()函数，演示如何调用代理查询天气，方便初学者验证代码。

def test_agent():
    initial_state = {
        "messages": [
            HumanMessage(content="北京今天天气如何？")
        ]
    }
    result = compiled_graph.run(initial_state)
    print("最终结果:", result["messages"][0])

if __name__ == "__main__":
    test_agent()

四、运行与测试：验证优化效果

1. 环境准备

• 安装依赖：

  pip install langchain langgraph sqlalchemy requests python-dotenv openai

• 创建.env文件，配置API密钥：

  API_KEY=your_openrouter_api_key
  NEWS_API_ORG=your_newsapi_key

2. 测试场景

• 场景1：查询天气
  输入：“北京今天天气如何？”
  输出：“北京的温度是16度，天气晴朗。”

• 场景2：插入用户信息
  输入：“我叫张三，年龄25，邮箱zhangsan@example.com，电话13812345678”
  代理会解析信息并调用insert_db工具，返回“数据已成功存储至MySQL数据库，用户ID：1”。

• 场景3：搜索实时新闻
  输入：“请告诉我最新的科技新闻”
  代理调用新闻API，返回第一条新闻的标题和内容。

五、总结：初学者必知的Agent开发要点

1. 工具设计原则：

   • 每个工具实现单一功能（如搜索、天气、数据库操作），通过Pydantic模型规范输入输出。
   • 工具需包含错误处理，返回结构化的成功/失败信息。

2. 状态管理核心：

   • 使用状态图（StateGraph）或流程引擎管理对话逻辑，明确“何时调用工具”和“何时直接回答”。
   • 通过LLM的结构化输出（如FinalResponse模型）判断是否需要工具调用。

3. 性能与健壮性：

   • 对外部服务（如API、数据库）添加超时、重试和异常捕获机制。
   • 合理使用连接池、缓存等优化手段，避免资源耗尽。

4. 代码可维护性：

   • 遵循PEP 8代码规范，添加详细注释和文档字符串。
   • 将配置（如API密钥、数据库地址）存储在环境变量中，避免硬编码。

六、扩展思考：进阶功能方向

• 多轮对话：支持追问用户获取缺失信息（如用户未提供邮箱时，询问“能否补充您的邮箱地址？”）。
• 工具扩展：添加更多工具（如计算器、翻译、文件操作等），丰富代理能力。
• 记忆管理：使用向量数据库（如Chroma）存储对话历史，实现上下文感知的长对话。

通过本文的案例和优化实践，初学者可以快速掌握智能代理的核心开发思路，并在此基础上构建更复杂的AI应用。完整代码如下：

import requests
import json
from typing import Union, Optional, Annotated
from typing import TypedDict, Annotated
import operator
from pydantic import BaseModel, Field
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_core.tools import tool
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, ToolMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
import os

# 加载环境变量
load_dotenv()

# 配置模型和API
MODEL_NAME = "openai/gpt-4o-mini"
BASE_URL = "https://openrouter.ai/api/v1"
OPENROUTER_API_KEY = os.getenv("API_KEY")
NEWS_API_KEY = os.getenv("NEWS_API_ORG")

# 数据库配置
DATABASE_URI = 'mysql+pymysql://root:Aa%40111111@localhost:3306/ai_agent?charset=utf8mb4'
engine = create_engine(DATABASE_URI, pool_size=5, max_overflow=10, pool_timeout=30)

# 创建基类
Base = declarative_base()

# 定义 User 模型
class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    age = Column(Integer)
    email = Column(String(100), nullable=False, unique=True)
    phone = Column(String(15))

# 创建表
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

# 工具模型定义
class SearchQuery(BaseModel):
    query: str = Field(description="用于网络查询的问题")

class WeatherLoc(BaseModel):
    location: str = Field(description="城市的地理位置名称")

class UserInfo(BaseModel):
    """提取的用户信息，如姓名、年龄、电子邮件和电话号码（如相关）"""
    name: str = Field(description="用户的姓名")
    age: Optional[int] = Field(description="用户的年龄")
    email: str = Field(description="用户的电子邮件地址")
    phone: Optional[str] = Field(description="用户的电话号码")

# 工具函数
@tool(args_schema=SearchQuery)
def fetch_real_time_info(query: str) -> str:
    """获取实时互联网信息"""
    url = "https://newsapi.org/v2/everything"
    params = {
        'apiKey': NEWS_API_KEY,
        'sortBy': 'publishedAt',
        'q': query
    }

    try:
        response = requests.get(url, params=params, timeout=10)
        response.raise_for_status()
        articles = response.json().get('articles')
        if articles:
            return json.dumps(articles[0], ensure_ascii=False, indent=2)
        else:
            return json.dumps({"error": "未找到相关结果"}, ensure_ascii=False)
    except requests.exceptions.RequestException as e:
        return json.dumps({"error": f"网络请求错误: {str(e)}"}, ensure_ascii=False)

@tool(args_schema=WeatherLoc)
def get_weather(location: str) -> str:
    """调用以获取当前天气,location参数请强制转成英语"""
    # 实际应用中可以替换为真实的天气API调用
    location = location.lower()
    weather_data = {
        "beijing": "北京的温度是16度，天气晴朗。",
        "shanghai": "上海的温度是20度，部分多云。"
    }
    return weather_data.get(location, "不好意思，并未查询到具体的天气信息。")

@tool(args_schema=UserInfo)
def insert_db(user_info: UserInfo) -> dict:
    """将用户信息插入数据库，必需参数为姓名、电子邮件"""
    session = Session()
    try:
        # 检查邮箱是否已存在
        existing_user = session.query(User).filter_by(email=user_info.email).first()
        if existing_user:
            return {"messages": [f"插入失败：邮箱地址 {user_info.email} 已存在"]}

        # 创建用户实例
        user = User(
            name=user_info.name,
            age=user_info.age,
            email=user_info.email,
            phone=user_info.phone
        )

        # 添加到会话并提交
        session.add(user)
        session.commit()
        return {"messages": [f"数据已成功存储至MySQL数据库，用户ID：{user.id}"]}
    except Exception as e:
        session.rollback()
        return {"messages": [f"数据存储失败，错误原因：{str(e)}"]}
    finally:
        session.close()

您暂时无权查看此隐藏内容！

# 简单测试函数
def test_agent():
    initial_state = {
        "messages": [
            HumanMessage(content="我想知道北京今天天气如何？")
        ]
    }

    result = compiled_graph.invoke(initial_state)
    print("智能助手回应:", result["messages"][-1])

if __name__ == "__main__":
    test_agent()