Agent调用工具入门指南:从零构建你的第一个智能代理

1次阅读
没有评论

共计 2926 个字符,预计需要花费 8 分钟才能阅读完成。

image.webp

技术背景:为什么需要 Agent 调用工具

传统的 API 调用是典型的请求 - 响应模式(Request-Response),就像点外卖——你下单后等待商家送餐。而 Agent 调用工具更像是雇佣了一位私人厨师(Agent),不仅能按菜单做菜,还会主动建议时令菜品、调整火候(主动决策)。这种模式特别适合需要持续交互、动态决策的场景,比如智能客服、自动化流程等。

Agent 调用工具入门指南:从零构建你的第一个智能代理

核心区别在于:

  • 响应式 vs 主动式:传统 API 被动等待指令,Agent 可基于环境主动触发工具
  • 单次交互 vs 会话管理:Agent 能维护上下文,实现多轮对话式调用
  • 刚性流程 vs 动态编排:工具间组合方式可运行时决定

环境准备:搭建你的第一个 Agent 实验室

  1. 安装 Python 3.8+(推荐使用 pyenv 管理多版本)

    pyenv install 3.8.12
    pyenv local 3.8.12

  2. 创建虚拟环境并安装核心库

    python -m venv agent_env
    source agent_env/bin/activate  # Linux/Mac
    pip install langchain openai tiktoken

  3. 环境变量配置(建议使用 .env 文件)

    # .env 文件示例
    OPENAI_API_KEY=sk-your-key-here
    TOOL_TIMEOUT=30  # 默认超时秒数

核心实现:构建会使用工具的智能体

工具注册机制

想象工具注册就像给厨师准备厨具——每件工具都需要明确:

  • 功能描述:让 Agent 知道什么时候该用这个工具
  • 参数规范:就像厨具使用说明书
  • 执行方法:具体的操作逻辑
from langchain.agents import tool

@tool  # 装饰器将普通函数转为工具
def search_recipe(query: str) -> str:
    """查询菜谱工具,输入食材名称返回推荐做法"""
    # 实际实现可能是调用外部 API 或本地数据库
    return f"找到 {query} 的 3 种做法:..."

调用流程时序(简化版)

  1. Agent 接收用户输入(” 如何做糖醋排骨?”)
  2. 决策阶段:根据输入和上下文选择合适工具
  3. 执行阶段:调用search_recipe("糖醋排骨")
  4. 结果处理:整理工具返回并生成自然语言响应

完整代码示例

from langchain.agents import initialize_agent, AgentType
from langchain.llms import OpenAI
import os
from dotenv import load_dotenv

load_dotenv()  # 加载.env 文件

llm = OpenAI(temperature=0)  # 低随机性保证稳定性
tools = [search_recipe]  # 注册所有工具

# 构建 Agent
chef_agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True  # 打印详细执行过程
)

# 带错误处理的调用示例
try:
    response = chef_agent.run("教我做红烧鱼")
    print(response)
except Exception as e:
    print(f"Agent 执行失败: {str(e)}")
    # 可添加重试逻辑或降级方案

生产环境生存指南

超时控制双保险

  1. 工具级别超时(底层保障)

    from functools import partial
    
    @tool
    def safe_search(query: str, timeout: int = int(os.getenv('TOOL_TIMEOUT'))) -> str:
        """带超时保护的查询"""
        try:
            # 使用 concurrent.futures 实现
            with ThreadPoolExecutor() as executor:
                future = executor.submit(real_search, query)
                return future.result(timeout=timeout)
        except TimeoutError:
            return "查询超时,请稍后再试"

  2. Agent 整体超时(全局控制)

    from langchain.agents import AgentExecutor
    
    executor = AgentExecutor.from_agent_and_tools(
        agent=chef_agent,
        tools=tools,
        max_execution_time=60  # 单位秒
    )

限流实现方案

令牌桶算法示例(使用 redis 实现):

import redis
from datetime import timedelta

r = redis.Redis()

def check_rate_limit(user_id: str) -> bool:
    key = f"rate_limit:{user_id}"
    current = r.incr(key)
    if current == 1:
        r.expire(key, timedelta(minutes=1))  # 每分钟重置
    return current <= 30  # 每分钟 30 次限制

日志记录三要素

  1. 记录完整对话上下文
  2. 标记工具调用耗时
  3. 保存错误堆栈信息

推荐结构:

{
  "timestamp": "2023-07-20T14:30:00Z",
  "session_id": "abc123",
  "input": "如何做佛跳墙",
  "tool_used": "search_recipe",
  "execution_time_ms": 1200,
  "error": null
}

新手避坑三大雷区

  1. 工具描述模糊
  2. 症状:Agent 频繁调用错误工具
  3. 修复:完善工具函数的 docstring,包含示例

  4. 未处理工具异常

  5. 症状:单个工具失败导致整个 Agent 崩溃
  6. 修复:所有工具函数内部添加 try-catch

  7. 无限循环调用

  8. 症状:Agent 不断重复调用同一工具
  9. 修复:设置 max_iterations 参数
    AgentExecutor(max_iterations=5)

扩展思考:工具优先级设计

当多个工具都匹配当前需求时,可以通过以下维度建立优先级:

  1. 精确匹配度:计算输入与工具描述的相似度
  2. 历史成功率:记录各工具过去调用表现
  3. 执行成本:API 费用或计算资源消耗

实现示例:

def select_tool(input_text, tools):
    scores = []
    for tool in tools:
        # 使用余弦相似度计算匹配分
        desc_score = cosine_similarity(embed(input_text), embed(tool.description))
        # 综合历史成功率(假设有记录)success_rate = tool_stats[tool.name]["success"]
        scores.append(desc_score * 0.7 + success_rate * 0.3)
    return tools[np.argmax(scores)]

学习资源宝库

记住,好的 Agent 就像训练有素的助手——给它清晰的工具说明,建立可靠的错误处理机制,再配合合理的限制条件,就能创造出惊人的自动化价值。

正文完
 0
评论(没有评论)