尧图建网站 尧图建网站 YAOTU WEB BUILD 免费咨询
ARTICLE DETAIL

资讯详情

深耕网站建设与建站编程的一线实战洞察。

AI助手开发实战:基于大语言模型的工具调用与Agent实现

AI助手开发实战:基于大语言模型的工具调用与Agent实现 这次我们来看一个AI助手开发项目重点是如何让一个只会“动嘴”的大语言模型通过“工具调用”能力真正“动手”去执行任务。项目标题“从零开发AI助手-Agent Tool Use 工具调用下篇”已经点明了核心为模型装上“手”。这意味着我们将构建一个能理解用户意图、自主选择并调用外部工具如查询时间、计算数学、搜索信息的智能体Agent而不仅仅是生成文本。这个项目的核心价值在于实践落地。它不空谈Agent概念而是聚焦于如何用代码实现一个具备工具调用能力的AI助手。对于开发者而言最关心的是需要什么环境代码结构如何如何定义工具模型如何与工具交互本文将围绕这些实际问题带你从环境搭建到功能验证完整走通一个支持“查时间”和“算算术”的AI助手开发流程。如果你正在学习AI应用开发希望将大模型的能力从对话扩展到执行具体操作或者想了解Function Calling/Tool Use背后的工程实现那么这篇文章正是你需要的。我们将重点关注实现方案、代码结构、接口设计以及本地测试的完整闭环。1. 核心能力速览在深入代码之前我们先通过下表快速了解这个AI助手项目的核心特性和要求能力项说明项目类型AI智能体Agent开发实践聚焦工具调用Tool Use/Function Calling核心功能1.意图理解解析用户自然语言指令如“现在几点”、“计算123456”。2.工具调度根据意图自动选择并调用预定义的工具函数。3.结果整合获取工具执行结果后组织成自然语言回复给用户。关键技术大语言模型LLM的Function Calling能力、JSON Schema定义工具、Agent决策循环开发环境Python 3.8 需要能调用具备工具调用能力的LLM API如OpenAI GPT系列、DeepSeek、智谱GLM等或本地模型硬件门槛无强制GPU要求。本项目主要依赖外部LLM API本地仅运行轻量的Agent框架代码普通CPU电脑即可运行。若使用本地部署的LLM则需满足对应模型的硬件要求。启动方式通过Python脚本直接启动通常是一个简单的python app.py或python agent_main.py。是否支持API是。项目本身可以封装为Web服务如使用FastAPI提供统一的对话接口内部处理工具调用逻辑。是否支持批量任务取决于架构设计。核心的Agent决策逻辑可以设计为支持异步或队列处理从而应对批量用户请求。适合场景1. 学习AI Agent开发原理。2. 构建可执行具体任务的个人助手或客服机器人。3. 作为更复杂Agent系统如结合RAG、工作流的基础模块。2. 适用场景与使用边界这个工具调用AI助手项目其价值在于将大语言模型的“认知”能力与外部工具的“执行”能力相结合。它非常适合以下场景教育学习作为理解Agent和Function Calling机制的绝佳实践项目代码结构清晰功能聚焦。原型验证快速验证某个业务场景如订单查询、数据统计、信息检索能否通过AI Agent自动化。效率工具开发个人桌面助手通过自然语言完成一系列操作如“查天气”、“发邮件”、“记笔记”。复杂系统基石作为更大智能体系统的“大脑”负责任务规划和工具调度后续可接入知识库RAG、长期记忆、多智能体协作等模块。需要注意的使用边界依赖LLM能力工具调用的准确性和可靠性很大程度上取决于底层大语言模型对Function Calling的支持好坏。模型需要准确理解用户指令并匹配到正确的工具。工具需预先定义AI助手只能调用开发者预先编写并注册好的工具函数。它无法创造或调用未知的工具。安全与权限工具可能执行文件操作、网络请求等。必须严格控制工具的执行权限避免被恶意指令利用例如删除文件、访问敏感API。在定义工具时要做好输入校验和权限隔离。非实时控制对于需要毫秒级响应的实时控制系统如工业机械臂控制基于LLM的Agent决策循环可能延迟过高不适合作为核心控制器。3. 环境准备与前置条件开始编码前请确保你的开发环境满足以下要求。1. 基础开发环境操作系统Windows 10/11, macOS, 或 Linux (如Ubuntu 20.04)均可。Python版本 3.8 或以上。推荐使用 3.10 兼容性最好。包管理工具pip已安装并更新至最新版。2. 大语言模型访问权限这是项目的核心依赖。你需要选择一个支持工具调用Function Calling的LLM服务。常见选项有OpenAI APIGPT-3.5-turbo或GPT-4系列对Function Calling支持非常成熟。国内大模型API如智谱AIChatGLM、百度文心一言、阿里通义千问、DeepSeek等。需查阅其最新官方文档确认是否开放了工具调用接口。本地部署模型如使用Ollama、LM Studio等工具部署的本地模型如Qwen、Llama等支持function calling的版本。这会引入GPU/显存要求需自行准备。3. 获取API密钥如使用云端API前往对应平台的官网注册账号并获取API Key。妥善保管不要直接提交到代码仓库。4. 代码编辑器或IDE推荐使用 VSCode、PyCharm 等它们对Python开发和调试支持良好。4. 安装部署与启动方式本项目不涉及复杂的服务部署核心是一个Python应用程序。我们按步骤搭建项目。第一步创建项目目录与虚拟环境为了避免包冲突强烈建议使用虚拟环境。# 创建项目文件夹 mkdir ai_assistant_agent cd ai_assistant_agent # 创建Python虚拟环境 (以venv为例) python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/macOS: source venv/bin/activate激活后命令行提示符前会出现(venv)标识。第二步安装核心依赖我们将使用openai库作为与LLM交互的客户端即使使用国内API其SDK通常也兼容OpenAI格式。同时安装requests用于可能的网络工具pydantic用于数据验证和JSON Schema生成。pip install openai requests pydantic如果你计划使用特定国产模型的官方SDK请根据其文档安装例如zhipuai、dashscope等。第三步项目文件结构一个清晰的项目结构有助于管理。创建如下文件ai_assistant_agent/ ├── tools/ # 工具模块目录 │ ├── __init__.py │ └── calculator.py # 计算器工具 │ └── time_query.py # 时间查询工具 ├── agents/ # 智能体模块目录 │ ├── __init__.py │ └── base_agent.py # 基础Agent类 ├── config.py # 配置文件存放API Key等 ├── main.py # 主程序入口 └── requirements.txt # 依赖列表第四步编写配置文件在config.py中安全地配置你的API密钥和模型信息。# config.py import os from dotenv import load_dotenv # 可选用于从.env文件加载 # 如果使用.env文件取消下面注释 # load_dotenv() class Config: # 使用环境变量或直接填写不推荐直接写死 # 例如OPENAI_API_KEY os.getenv(OPENAI_API_KEY) LLM_API_KEY your_api_key_here # 请替换为你的真实Key LLM_BASE_URL https://api.openai.com/v1 # 如果是其他模型需修改此地址 LLM_MODEL gpt-3.5-turbo # 指定使用的模型名称 config Config()重要将“your_api_key_here”替换为你的真实API Key。更安全的方式是使用os.getenv从系统环境变量读取。第五步启动与测试完成后续的代码编写后启动方式非常简单# 确保在项目根目录且虚拟环境已激活 python main.py程序将启动一个简单的交互式命令行界面CLI等待你输入问题。5. 功能测试与效果验证现在我们来构建核心功能并进行测试。我们将实现两个基础工具get_current_time获取当前时间和calculate执行数学计算。5.1 定义工具给模型装上“手”首先在tools目录下创建工具。我们使用Pydantic模型来定义工具的参数Schema这能自动生成符合LLM要求的JSON Schema。工具一时间查询工具 (tools/time_query.py)# tools/time_query.py from datetime import datetime from pydantic import BaseModel, Field from typing import Literal # 定义工具的输入参数模型 class TimeQueryInput(BaseModel): timezone: str Field( defaultAsia/Shanghai, descriptionThe timezone to get the current time for, e.g., Asia/Shanghai, America/New_York. ) format: Literal[full, time_only, date_only] Field( defaultfull, descriptionThe format of the output. full for date and time, time_only for time only, date_only for date only. ) # 工具函数本身 def get_current_time(timezone: str Asia/Shanghai, format: str full) - str: Get the current time for a specified timezone. # 简化处理这里使用本地时间实际应使用pytz库处理时区 # 为简化示例我们忽略时区转换直接返回本地时间 now datetime.now() if format time_only: return now.strftime(%H:%M:%S) elif format date_only: return now.strftime(%Y-%m-%d) else: # full return now.strftime(%Y-%m-%d %H:%M:%S) # 工具元信息供Agent注册和使用 time_tool { function: get_current_time, # 函数对象 name: get_current_time, description: Get the current date and/or time., parameters_schema: TimeQueryInput.model_json_schema(), # Pydantic自动生成Schema }工具二计算器工具 (tools/calculator.py)# tools/calculator.py import math from pydantic import BaseModel, Field from typing import Literal class CalculateInput(BaseModel): expression: str Field(descriptionThe mathematical expression to evaluate, e.g., 3 5 * (2 - 1).) # 可以扩展更多参数如精度 # precision: int Field(default2, ge0, le10, descriptionDecimal precision for the result.) def calculate(expression: str) - str: Evaluate a mathematical expression. WARNING: Using eval() is dangerous. This is for demonstration only. In production, use a safe evaluator like asteval or implement a parser. # 安全警告此处使用eval仅为演示实际项目必须使用安全的表达式求值库 try: # 极其有限的安全检查不完善 allowed_chars set(0123456789-*/(). ) if not all(c in allowed_chars for c in expression): return Error: Expression contains unsafe characters. result eval(expression, {__builtins__: {}}, {}) return str(result) except Exception as e: return fError evaluating expression: {e} calculator_tool { function: calculate, name: calculate, description: Evaluate a mathematical expression. Supports basic operators: , -, *, /, (), and numbers., parameters_schema: CalculateInput.model_json_schema(), }5.2 构建基础Agent (agents/base_agent.py)Agent的核心是决策循环接收用户输入 - LLM决定是否调用工具及调用哪个 - 执行工具 - 将结果返回给LLM - LLM生成最终回复。# agents/base_agent.py import json from openai import OpenAI from config import config from typing import List, Dict, Any class BaseAgent: def __init__(self, tools: List[Dict[str, Any]]): 初始化Agent。 :param tools: 工具列表每个工具是一个包含name, description, parameters_schema, function的字典。 self.client OpenAI( api_keyconfig.LLM_API_KEY, base_urlconfig.LLM_BASE_URL, ) self.model config.LLM_MODEL self.tools tools # 将工具函数映射到名称便于调用 self.tool_function_map {tool[name]: tool[function] for tool in tools} # 构建供LLM识别的工具描述列表 self.llm_tools [ { type: function, function: { name: tool[name], description: tool[description], parameters: tool[parameters_schema], } } for tool in tools ] def run(self, user_input: str) - str: 运行Agent的主要循环。 messages [{role: user, content: user_input}] # 第一步LLM决定是否需要调用工具 response self.client.chat.completions.create( modelself.model, messagesmessages, toolsself.llm_tools, tool_choiceauto, # 让模型自动决定 ) response_message response.choices[0].message tool_calls response_message.tool_calls # 第二步如果有工具调用则执行 if tool_calls: messages.append(response_message) # 将包含工具调用的消息加入历史 for tool_call in tool_calls: function_name tool_call.function.name function_to_call self.tool_function_map.get(function_name) if function_to_call: # 解析模型传入的参数 function_args json.loads(tool_call.function.arguments) # 执行工具函数 function_response function_to_call(**function_args) # 将工具执行结果返回给LLM messages.append({ role: tool, tool_call_id: tool_call.id, content: str(function_response), # 确保是字符串 name: function_name, }) # 第三步将工具结果返回给LLM让它生成最终回复 second_response self.client.chat.completions.create( modelself.model, messagesmessages, ) final_message second_response.choices[0].message.content return final_message else: # 没有工具调用直接返回模型回复 return response_message.content or No response generated.5.3 主程序与交互测试 (main.py)最后我们将所有部分串联起来。# main.py from tools.time_query import time_tool from tools.calculator import calculator_tool from agents.base_agent import BaseAgent def main(): # 1. 注册所有可用工具 available_tools [time_tool, calculator_tool] # 2. 初始化Agent print(正在初始化AI助手工具调用版...) agent BaseAgent(toolsavailable_tools) print(初始化完成输入您的问题输入 quit 或 exit 退出\n) # 3. 启动交互循环 while True: try: user_input input( 您: ).strip() if user_input.lower() in [quit, exit, q]: print(再见) break if not user_input: continue # 4. 运行Agent print(AI助手正在思考...) response agent.run(user_input) print(f 助手: {response}\n) except KeyboardInterrupt: print(\n程序被中断。) break except Exception as e: print(f发生错误: {e}\n) if __name__ __main__: main()5.4 效果验证现在启动程序进行测试。python main.py你应该看到类似以下的交互过程测试用例1查询时间 您: 现在几点了 AI助手正在思考... 助手: 现在是2024-05-27 15:30:45。验证点模型正确识别了“现在几点了”的意图并调用了get_current_time工具返回了格式化的时间。测试用例2数学计算 您: 请计算一下 (15 7) * 3 等于多少 AI助手正在思考... 助手: (15 7) * 3 的计算结果是 66。验证点模型识别出数学计算意图调用了calculate工具并正确执行了运算。测试用例3混合意图需模型自主规划 您: 先告诉我现在的时间然后计算从今天到2024年12月31日还有多少天。 AI助手正在思考... 助手: 现在是2024-05-27 15:31:20。接下来计算到2024年12月31日的天数... 计算结果约为218天。验证点模型成功理解了一个包含两个子任务查询时间、计算日期差的复杂指令。它需要先调用时间工具再基于结果调用计算工具示例中calculate工具可能不支持日期计算需扩展。这展示了Agent的任务分解潜力。测试用例4无需工具调用的普通对话 您: 你好请介绍一下你自己。 AI助手正在思考... 助手: 你好我是一个具备工具调用能力的AI助手。除了和你聊天我还可以帮你查询当前时间、进行简单的数学运算等。有什么可以帮你的吗验证点对于不需要工具调用的问候或闲聊模型不会触发工具调用直接生成回复。这证明了工具调用的“按需”特性。6. 接口API与批量任务虽然我们构建了一个CLI程序但将其改造为Web API服务是迈向实用的关键一步。这样其他应用或前端界面就能方便地调用。6.1 使用FastAPI构建Web服务安装FastAPI和Uvicornpip install fastapi uvicorn创建api_server.py# api_server.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional from tools.time_query import time_tool from tools.calculator import calculator_tool from agents.base_agent import BaseAgent import uvicorn app FastAPI(titleAI助手工具调用API) # 全局Agent实例 available_tools [time_tool, calculator_tool] agent BaseAgent(toolsavailable_tools) class ChatRequest(BaseModel): message: str session_id: Optional[str] None # 可用于多轮对话会话管理 class ChatResponse(BaseModel): reply: str session_id: Optional[str] None app.post(/chat, response_modelChatResponse) async def chat_with_agent(request: ChatRequest): 与AI助手对话的端点。 try: user_input request.message if not user_input: raise HTTPException(status_code400, detailMessage cannot be empty.) # 调用Agent核心逻辑 response agent.run(user_input) return ChatResponse(replyresponse, session_idrequest.session_id) except Exception as e: raise HTTPException(status_code500, detailfInternal server error: {str(e)}) app.get(/health) async def health_check(): return {status: healthy} if __name__ __main__: # 启动服务默认在 http://127.0.0.1:8000 uvicorn.run(app, host0.0.0.0, port8000)6.2 启动与调用API服务启动服务python api_server.py使用curl或Pythonrequests库测试API使用curl测试curl -X POST http://127.0.0.1:8000/chat \ -H Content-Type: application/json \ -d {message: 现在北京时间是多少}使用Python测试import requests import json url http://127.0.0.1:8000/chat payload { message: 计算 99 除以 3 再加上 15 等于多少 } headers { Content-Type: application/json } response requests.post(url, jsonpayload, headersheaders, timeout30) if response.status_code 200: result response.json() print(f助手回复: {result[reply]}) else: print(f请求失败: {response.status_code}, {response.text})6.3 支持批量任务对于批量处理例如有一个包含大量问题的文件我们可以设计一个简单的异步队列或直接循环调用。批量处理脚本示例 (batch_process.py):# batch_process.py import asyncio import aiohttp import json from typing import List async def process_one(session: aiohttp.ClientSession, question: str, api_url: str) - dict: 处理单个问题 payload {message: question} try: async with session.post(api_url, jsonpayload) as resp: if resp.status 200: result await resp.json() return {question: question, answer: result.get(reply), status: success} else: return {question: question, answer: None, status: ferror_{resp.status}} except Exception as e: return {question: question, answer: None, status: fexception_{str(e)}} async def batch_process(questions: List[str], api_url: str http://127.0.0.1:8000/chat, concurrency: int 5): 批量处理问题列表 connector aiohttp.TCPConnector(limitconcurrency) timeout aiohttp.ClientTimeout(total60) async with aiohttp.ClientSession(connectorconnector, timeouttimeout) as session: tasks [process_one(session, q, api_url) for q in questions] results await asyncio.gather(*tasks) return results if __name__ __main__: # 示例问题列表 questions [ 现在几点了, 123乘以456等于多少, 今天的日期是什么, 计算圆的面积假设半径是5。, ] # 运行批量处理 loop asyncio.get_event_loop() all_results loop.run_until_complete(batch_process(questions)) for res in all_results: print(f问题: {res[question]}) print(f状态: {res[status]}) print(f回答: {res[answer]}\n)这个脚本使用aiohttp进行异步HTTP请求可以并发处理多个问题显著提高批量任务效率。你需要根据实际API的响应时间和服务器负载调整concurrency并发数参数。7. 资源占用与性能观察由于本项目核心是Agent逻辑编排和API调用本地资源占用很低。CPU/内存占用运行Agent框架和FastAPI服务的Python进程本身消耗很小通常CPU使用率在1-5%之间内存占用在100-300MB左右主要取决于Python解释器和加载的库。网络延迟性能瓶颈主要在于调用大语言模型API的网络延迟。从发送请求到收到模型回复通常需要1到10秒不等取决于模型复杂度、网络状况和API服务方的负载。Token消耗工具调用会增加对话的Token消耗。一次完整的工具调用流程用户输入 模型思考是否调用工具 工具调用请求 工具执行结果 模型生成最终回复通常比简单对话消耗更多Token需关注API成本。本地模型部署如果选择在本地部署支持工具调用的LLM如通过Ollama则资源占用将完全由该本地模型决定。一个7B参数量的模型在GPU上推理可能需要6-8GB显存在CPU上推理则会占用大量内存和CPU时间。性能优化建议缓存工具结果对于频繁查询且结果变化不快的工具如某些信息查询可以添加缓存机制。异步处理如batch_process.py所示对于批量请求或Web服务使用异步框架如FastAPI async/await可以高效处理并发。超时与重试在调用外部API或工具时务必设置合理的超时时间并考虑实现重试逻辑以提高鲁棒性。监控Token使用在Agent类中添加逻辑统计每次交互的输入/输出Token数便于成本分析和优化提示词。8. 常见问题与排查方法在开发和使用过程中你可能会遇到以下问题问题现象可能原因排查方式解决方案启动报错ModuleNotFoundError依赖包未安装或虚拟环境未激活。检查命令行前缀是否有(venv)运行pip list查看是否安装了openai,pydantic等。激活虚拟环境运行pip install -r requirements.txt。API调用失败认证错误API Key错误、过期或未正确设置。检查config.py中的LLM_API_KEY或环境变量OPENAI_API_KEY。尝试用curl直接调用API验证。使用正确的API Key确保其有足够的余额和权限。模型不调用工具直接回答1. 模型能力不足。2. 工具描述description不清晰。3. 用户指令模糊。检查LLM返回的原始消息看是否有tool_calls字段。简化指令测试如直接说“调用计算器计算11”。1. 尝试更强大的模型如GPT-4。2. 优化工具描述使其更精准。3. 在系统提示词system message中明确要求模型使用工具。工具调用参数解析错误模型生成的参数JSON格式错误或与Pydantic Schema不匹配。打印tool_call.function.arguments查看模型输出的原始参数。确保Pydantic Schema定义准确。可在Agent代码中添加更健壮的JSON解析和错误处理。工具函数执行出错工具函数内部逻辑有Bug或传入参数类型不对。在工具函数内添加print语句或使用日志记录输入参数。检查异常堆栈。修复工具函数代码。确保函数参数类型与Schema定义一致。Web服务启动后无法访问端口被占用或防火墙阻止。检查uvicorn启动日志是否有错误。用netstat -ano | findstr :8000(Win)或lsof -i:8000(Mac/Linux)查看端口。更换端口如port8001。确保防火墙允许该端口的入站连接。批量任务速度慢同步顺序调用网络延迟累积。观察任务处理日志看是否是一个接一个执行。改用异步并发处理如使用asyncio和aiohttp。calculate工具安全风险使用了不安全的eval()函数。审查tools/calculator.py中的calculate函数。必须替换为安全的表达式求值库如asteval或自己实现四则运算解析器。9. 最佳实践与使用建议为了让你的AI助手更健壮、安全、易用请遵循以下建议工具设计原则单一职责每个工具只做一件事并做好。这有助于模型准确理解和调用。描述清晰工具的name和description要精确、无歧义这是模型选择工具的主要依据。Schema严谨使用Pydantic严格定义输入参数的类型、默认值和约束这能自动生成高质量的JSON Schema并做输入验证。安全第一永远不要信任来自LLM的未经验证的输入。在工具函数内部对输入进行严格的校验、清理和权限检查。特别是执行系统命令、文件操作、网络请求的工具。Agent提示工程可以在发送给模型的messages列表开头加入一个system角色消息明确指导模型的行为。例如“你是一个有帮助的助手可以调用工具来回答问题。当你需要获取实时信息或进行计算时请务必调用相应的工具。”错误处理与日志在Agent循环和工具函数中全面添加try...except块捕获并记录异常给用户友好的错误提示而不是让程序崩溃。引入日志库如logging记录详细的运行日志包括用户输入、模型决策、工具调用、结果输出等便于调试和审计。扩展方向更多工具逐步添加如天气查询、网页搜索、邮件发送、数据库查询等实用工具。记忆与多轮对话为Agent添加会话记忆使其能处理上下文相关的多轮对话。工具组合与规划实现更复杂的Agent能够自动将复杂问题分解为多个工具调用步骤规划。前端界面使用Gradio、Streamlit或Web前端框架为你的助手构建一个图形化界面。合规与伦理明确告知用户正在与AI交互并且AI可能会调用外部工具。对于处理用户数据的工具必须遵守隐私政策。避免开发可能用于欺诈、骚扰或产生有害内容的工具。通过这个项目你不仅实现了一个能“查时间算算术”的AI助手更掌握了一套让大模型“动手做事”的核心方法论。从工具定义、Agent决策循环到API服务化这套模式是构建更复杂、更实用AI应用的基础。建议从这个小项目出发不断迭代添加新工具优化提示词最终打造出真正能提升效率的个人智能工作伙伴。
返回列表