这次我们来看一个关于 AI Agent 与 Skill 开发的实战教程。如果你正在寻找如何将大语言模型LLM从一个“聊天机器人”升级为能自主调用工具、执行复杂工作流的“智能体”那么这篇文章就是为你准备的。我们将聚焦于 Agent 的核心概念、Skill 的设计与实现、以及如何构建一个可运行的 Multi-Agent 系统整个过程强调可落地、可验证。这个教程的核心不是空谈理论而是提供一套从零到一的实践路径。我们将重点关注几个关键问题如何定义和实现一个 Skill如何让 Agent 理解并调用 Tool如何设计 Workflow 来串联多个 Agent 的协作以及如何在一个本地或云端环境中快速搭建并测试这套系统。对于开发者而言理解这些是构建下一代 AI 应用的基础。本文适合有一定 Python 基础并对大模型应用开发感兴趣的读者。无论你是想为现有产品增加 AI 能力还是探索 Multi-Agent 系统的可能性都能从中获得可直接上手的代码示例和架构思路。接下来我们将从核心概念速览开始逐步深入到环境搭建、代码实现、效果验证和常见问题排查。1. 核心能力速览在深入细节之前我们先通过一个表格快速了解本教程所涵盖的核心技术栈和能力边界这有助于你判断是否要继续深入。能力项说明与涵盖范围技术焦点AI Agent智能体架构、Skill技能开发、Tool工具调用、Workflow工作流编排、Multi-Agent多智能体协作。核心组件大语言模型LLM作为“大脑” Skill/Tool 作为“手脚” Workflow 引擎作为“调度中心”。实践门槛需要基本的 Python 编程能力理解 HTTP API 调用和 JSON 数据处理。无需高端 GPU大部分逻辑测试可在 CPU 上完成。模型依赖可对接 OpenAI API、国内大模型 API如文心一言、通义千问或本地部署的开源模型需自行搭建 API 服务。部署方式提供基于 FastAPI 的轻量级 Web 服务示例支持一键启动方便进行接口测试和集成。关键产出可运行的 Agent 原型具备自定义 Skill、自动 Tool 选择与调用、以及简单顺序工作流的能力。适合场景自动化客服、智能数据分析助手、跨平台信息聚合机器人、内部业务流程自动化等。2. 适用场景与使用边界AI Agent 系统并非万能明确其适用场景和边界是成功实施的第一步。它非常适合解决以下类型的问题流程固定但步骤繁琐的任务例如从一封客户邮件中提取关键信息查询内部数据库生成回复草稿并发送通知。这个流程涉及自然语言理解、数据查询和内容生成可以由一个或多个 Agent 协作完成。需要连接多个外部工具或 API 的服务例如一个旅行规划 Agent需要调用天气 API、航班查询接口、酒店预订系统和地图服务最后生成一份整合报告。7x24小时在线的智能交互接口作为客服、导购或信息查询的第一道关口能够处理常见问题并在复杂情况下无缝转接人工。探索性数据分析与报告生成给定一个数据集和分析目标Agent 可以自动调用不同的分析工具如统计、可视化并组织成一份初步分析报告。当前技术的局限性使用边界复杂逻辑与深度推理对于需要极深领域知识或复杂逻辑链推理的任务如法律判决、医疗诊断现有 Agent 系统更多是辅助角色不能完全替代专家。高实时性与绝对精度要求在金融交易、工业控制等对实时性和精度要求极高的场景Agent 的决策延迟和不可预测性可能带来风险。完全开放域的创造性工作虽然能辅助创作但生成高度原创且连贯的艺术作品、长篇小说等仍需人类主导。安全与合规红线Agent 必须被约束在明确的工具和知识范围内运行严禁让其自行访问网络搜索或执行未经审核的代码以防产生有害内容或行为。重要合规提醒在开发涉及用户数据、隐私信息或内容生成的 Agent 时必须确保数据获取和使用获得明确授权。生成的内容符合法律法规和平台规范。系统设计包含审核机制和人工干预入口。3. 环境准备与前置条件为了顺利跟进后续的实战部分请确保你的开发环境满足以下基本要求。这是一个通用清单具体版本可能因所选框架而异。操作系统Windows 10/11 macOS 或 Linux如 Ubuntu 20.04。本教程示例代码在主流系统上均可运行。Python 环境推荐使用 Python 3.8 至 3.11 版本。这是大多数 AI 框架和网络库稳定支持的版本区间。使用python --version检查版本。强烈建议使用venv或conda创建独立的虚拟环境避免包冲突。包管理工具pip版本需保持较新。代码编辑器或 IDEVSCode、PyCharm 或任何你熟悉的编辑器。网络访问由于需要调用大模型 API除非使用本地模型请确保你的网络环境可以稳定访问相应的 API 服务提供商如 OpenAI、Azure OpenAI 或国内大模型平台。API 密钥准备一个可用的大模型 API 密钥。这是驱动 Agent“大脑”的燃料。你可以从 OpenAI、文心一言、通义千问、智谱 AI 等平台获取。请妥善保管你的密钥不要将其提交到代码仓库。4. 安装部署与启动方式我们将构建一个基于 FastAPI 的简易 Agent 服务器。它结构清晰易于扩展适合学习和原型开发。第一步创建项目并安装依赖在你的工作目录下执行以下命令# 1. 创建项目目录并进入 mkdir ai-agent-tutorial cd ai-agent-tutorial # 2. 创建虚拟环境以 venv 为例 python -m venv venv # 3. 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/macOS: source venv/bin/activate # 4. 创建依赖文件 requirements.txt并填入以下内容 # requirements.txt fastapi0.104.0 uvicorn[standard]0.24.0 openai1.0.0 # 如果你使用 OpenAI 官方库 requests2.31.0 pydantic2.0.0 python-dotenv1.0.0 # 用于管理环境变量 # 5. 安装依赖 pip install -r requirements.txt第二步组织项目结构一个清晰的结构有助于管理复杂的 Agent 系统。建议按如下方式组织文件ai-agent-tutorial/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 应用主入口 │ ├── agents/ # 存放不同 Agent 的定义 │ │ ├── __init__.py │ │ └── planner_agent.py │ ├── skills/ # 存放所有 Skill 的实现 │ │ ├── __init__.py │ │ ├── base_skill.py │ │ ├── calculator_skill.py │ │ └── weather_skill.py │ ├── tools/ # 存放具体的工具函数可被 Skill 调用 │ │ ├── __init__.py │ │ └── weather_tool.py │ ├── workflows/ # 存放工作流定义 │ │ ├── __init__.py │ │ └── simple_sequential.py │ └── config.py # 配置文件 ├── .env # 环境变量文件务必加入 .gitignore ├── requirements.txt └── README.md第三步编写核心代码与启动服务这里给出最简化的核心代码示例展示 Agent、Skill、Tool 如何联动。首先创建.env文件来存储你的 API 密钥# .env OPENAI_API_KEYsk-your-openai-api-key-here # 或者使用国内模型例如 DASHSCOPE_API_KEYsk-your-dashscope-api-key-here # 阿里通义千问接着实现一个基础 Skill 和 Tool。app/tools/weather_tool.pyimport requests from pydantic import BaseModel class WeatherQuery(BaseModel): city: str def get_weather(query: WeatherQuery) - str: 模拟获取天气的工具函数。 实际应用中这里应调用真实的天气API如和风天气、OpenWeatherMap等。 此处返回模拟数据。 # 模拟API调用延迟 import time time.sleep(0.5) # 返回模拟结果 return f{query.city}的天气晴温度 22°C湿度 65%东南风 2级。然后创建一个使用该 Tool 的 Skill。app/skills/weather_skill.pyfrom app.skills.base_skill import BaseSkill from app.tools.weather_tool import get_weather, WeatherQuery from pydantic import Field class WeatherSkill(BaseSkill): 天气查询技能 name: str weather_query description: str 根据城市名称查询该城市的实时天气情况。 city: str Field(..., description要查询天气的城市名称例如北京、上海) def execute(self): 执行技能的核心逻辑 query WeatherQuery(cityself.city) result get_weather(query) return {skill: self.name, result: result, input: {city: self.city}}Skill 的基类app/skills/base_skill.py可以这样定义from pydantic import BaseModel from abc import ABC, abstractmethod class BaseSkill(BaseModel, ABC): 所有技能的基类 name: str description: str abstractmethod def execute(self): 执行技能必须由子类实现 pass现在创建一个简单的 Agent它能“思考”并决定使用哪个 Skill。app/agents/planner_agent.pyimport os from openai import OpenAI from app.skills.weather_skill import WeatherSkill from app.skills.calculator_skill import CalculatorSkill # 假设你已实现计算器技能 class PlannerAgent: def __init__(self): # 初始化可用的技能列表 self.available_skills { weather_query: WeatherSkill, calculator: CalculatorSkill, } # 初始化 LLM 客户端这里以 OpenAI 为例 self.client OpenAI(api_keyos.getenv(OPENAI_API_KEY)) def plan_and_execute(self, user_query: str): 1. 分析用户查询规划需要使用的技能。 2. 执行技能并返回结果。 # 步骤1让 LLM 分析查询并选择技能 system_prompt f 你是一个任务规划助手。请根据用户查询从以下技能中选择最合适的一个。 技能列表 {self._format_skills_for_prompt()} 请只返回技能的名称不要返回其他任何内容。 response self.client.chat.completions.create( modelgpt-3.5-turbo, messages[ {role: system, content: system_prompt}, {role: user, content: user_query} ], temperature0, ) chosen_skill_name response.choices[0].message.content.strip() # 步骤2根据技能名称提取参数并实例化技能 skill_class self.available_skills.get(chosen_skill_name) if not skill_class: return {error: f未找到技能{chosen_skill_name}} # 步骤3让 LLM 提取技能所需参数简化版实际可用更复杂的参数提取 param_prompt f 用户查询是“{user_query}” 需要调用的技能是{chosen_skill_name} ({skill_class.description}) 请根据技能描述和用户查询提取出技能所需的参数并以JSON格式返回。 例如对于天气查询技能参数是 {{city: 北京}} param_response self.client.chat.completions.create( modelgpt-3.5-turbo, messages[{role: user, content: param_prompt}], temperature0, response_format{type: json_object} ) import json params json.loads(param_response.choices[0].message.content) # 步骤4实例化技能并执行 skill_instance skill_class(**params) result skill_instance.execute() return result def _format_skills_for_prompt(self): return \n.join([f- {name}: {cls.description} for name, cls in self.available_skills.items()])最后创建 FastAPI 主应用。app/main.pyfrom fastapi import FastAPI, HTTPException from pydantic import BaseModel from app.agents.planner_agent import PlannerAgent app FastAPI(titleAI Agent 演示服务) agent PlannerAgent() class QueryRequest(BaseModel): query: str app.post(/agent/query) async def handle_agent_query(request: QueryRequest): 处理用户查询的接口 try: result agent.plan_and_execute(request.query) return {success: True, data: result} except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.get(/health) async def health_check(): return {status: healthy}第四步启动服务在项目根目录下运行以下命令启动 FastAPI 服务uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload启动成功后你将看到类似Uvicorn running on http://0.0.0.0:8000的输出。现在你的第一个 AI Agent 服务已经运行在本地 8000 端口。5. 功能测试与效果验证服务启动后我们需要通过实际调用来验证 Agent 的核心能力理解意图、选择技能、执行工具、返回结果。5.1 基础技能调用测试我们将使用curl或 Pythonrequests库来测试接口。测试用例1天气查询# 使用 curl 测试 curl -X POST http://127.0.0.1:8000/agent/query \ -H Content-Type: application/json \ -d {query: 今天北京的天气怎么样}预期成功响应{ success: true, data: { skill: weather_query, result: 北京的天气晴温度 22°C湿度 65%东南风 2级。, input: { city: 北京 } } }验证点Agent 正确识别了用户意图为“天气查询”。成功提取了参数{city: 北京}。调用了get_weather工具函数并返回了模拟的天气结果。测试用例2数学计算假设已实现CalculatorSkillcurl -X POST http://127.0.0.1:8000/agent/query \ -H Content-Type: application/json \ -d {query: 请计算 125 乘以 38 等于多少}预期成功响应应包含技能名calculator和计算结果4750。5.2 多轮对话与上下文测试一个真正的 Agent 可能需要处理多轮对话。我们可以在PlannerAgent中引入简单的上下文记忆。这里展示一个扩展思路在PlannerAgent类中增加一个上下文存储例如一个列表并在每次交互时将历史对话作为上下文传递给 LLM。这需要修改plan_and_execute方法接收一个conversation_history参数并将其拼接到给 LLM 的提示词中。测试方法连续发送两条相关的查询如“北京天气如何” - Agent 回复天气。“那上海呢” - Agent 应能理解“上海”指的是“上海的天气”并调用天气查询技能。这个测试验证了 Agent 是否具备基础的上下文理解能力。5.3 技能选择准确性测试设计一些边界或模糊的查询测试 Agent 选择技能的准确性。查询“帮我订一张从北京到上海的机票。”期望由于我们未实现订票技能Agent 应返回错误或表示无法处理。我们的示例中会返回{error: 未找到技能xxx}。在实际系统中应设计一个“兜底”技能如通用问答或转人工。查询“今天气温多少度”未指定城市期望Agent 应能通过 LLM 的推理或追问机制补全参数例如根据上下文或默认城市。我们的简化示例可能失败这指出了系统需要改进的地方参数提取的鲁棒性。5.4 工作流Workflow测试Workflow 用于编排多个 Skill 的执行顺序。例如一个“出行报告”工作流可能依次执行查询天气-查询航班-生成摘要。创建一个简单的工作流app/workflows/simple_sequential.pyclass SimpleSequentialWorkflow: def __init__(self, skill_list): self.skills skill_list # skill_list 是技能实例的列表 def run(self, initial_input): result {} current_input initial_input for skill in self.skills: # 将上一步的输出作为下一步的输入这里做了简化 skill_result skill.execute() result[skill.name] skill_result # 在实际中可能需要更复杂的输入输出映射 current_input skill_result return result然后在 API 中新增一个端点来触发这个工作流。通过测试验证多个技能能否按预定顺序执行并将结果汇总。6. 接口 API 与批量任务我们的 Agent 服务通过 HTTP API 暴露这使其易于与前端、移动端或其他后端服务集成。同时也可以轻松扩展为处理批量任务。6.1 核心 API 接口说明目前我们只有一个核心接口端点POST /agent/query功能接收自然语言查询返回技能执行结果。请求体{ query: 用户输入的自然语言问题 }响应体{ success: true/false, data: { ... } // 技能执行结果或错误信息 }6.2 使用 Python 调用 API以下是一个调用示例你可以将其集成到你的自动化脚本中import requests import json def call_agent_service(query_text, api_urlhttp://127.0.0.1:8000/agent/query): 调用本地 Agent 服务 payload {query: query_text} headers {Content-Type: application/json} try: response requests.post(api_url, jsonpayload, headersheaders, timeout30) response.raise_for_status() # 检查HTTP错误 return response.json() except requests.exceptions.RequestException as e: return {success: False, error: f网络请求失败: {e}} except json.JSONDecodeError as e: return {success: False, error: f响应解析失败: {e}} # 示例调用 if __name__ __main__: result call_agent_service(今天北京的天气怎么样) print(json.dumps(result, indent2, ensure_asciiFalse))6.3 批量任务处理对于需要处理大量相似查询的场景如批量处理客服日志、分析用户反馈可以构建一个简单的批量任务处理器。创建一个脚本batch_processor.pyimport concurrent.futures from call_agent_service import call_agent_service # 导入上面的函数 def process_batch(queries, max_workers5): 并发处理一批查询。 :param queries: 查询字符串列表 :param max_workers: 最大并发线程数 :return: 结果列表顺序与输入对应 results [] with concurrent.futures.ThreadPoolExecutor(max_workersmax_workers) as executor: # 提交所有任务 future_to_query {executor.submit(call_agent_service, q): q for q in queries} # 按完成顺序收集结果 for future in concurrent.futures.as_completed(future_to_query): query future_to_query[future] try: result future.result() results.append((query, result)) except Exception as exc: results.append((query, {success: False, error: str(exc)})) # 按原始顺序排序如果需要 # results.sort(keylambda x: queries.index(x[0])) return results if __name__ __main__: task_list [ 北京天气, 计算 991, 上海明天的天气, 123乘以456是多少 ] batch_results process_batch(task_list) for query, resp in batch_results: print(f查询: {query}) print(f结果: {resp}\n)注意事项速率限制如果调用的是第三方大模型 API如 OpenAI请注意其速率限制RPM/TPM在批量任务中需要加入延迟或使用更高级的队列管理。错误处理批量任务中必须对每个子任务进行独立的错误捕获和记录避免单个任务失败导致整个批次中断。资源管理并发数 (max_workers) 不宜设置过高以免压垮本地服务或触发 API 限流。7. 资源占用与性能观察由于我们的示例 Agent 核心是调用远程大模型 API因此本地资源占用主要集中在 Web 服务FastAPI和少量的内存上。性能瓶颈主要在网络 I/O 和大模型 API 的响应时间。7.1 本地服务资源占用CPU/内存运行uvicorn进程本身占用极低。你可以使用系统任务管理器或htop命令观察通常内存占用在 100MB 以内CPU 使用率接近 0%空闲时。网络服务监听本地端口如 8000仅处理内部或局域网请求带宽消耗可忽略。7.2 性能关键点与优化大模型 API 延迟这是最主要的耗时环节。一次完整的plan_and_execute调用可能包含 2 次 LLM 交互规划参数提取。优化方法使用更快的模型如gpt-3.5-turbo比gpt-4快得多。优化提示词Prompt清晰、简洁的提示词能减少模型的“思考”时间并提高输出准确性。设置超时与重试在调用openai库或requests时合理设置timeout参数并实现简单的重试逻辑针对网络波动或 API 临时错误。技能执行效率如果 Skill 内部需要调用其他慢速 API如爬虫、复杂计算也会成为瓶颈。异步化将 FastAPI 的端点、Skill 的execute方法改为async def并使用httpx等异步 HTTP 客户端可以大幅提升高并发下的吞吐量。缓存对频繁查询且结果变化不频繁的数据如天气信息可缓存10分钟引入缓存机制如functools.lru_cache或 Redis。监控与观察使用uvicorn的--reload参数便于开发生产环境应移除。使用time模块在代码中关键位置打点记录耗时。考虑使用像Prometheus和Grafana这样的监控套件来收集服务的 QPS、延迟、错误率等指标。7.3 扩展为本地模型部署如果你想完全本地化避免网络延迟和 API 费用可以将 LLM 部分替换为本地部署的开源模型如 Qwen、ChatGLM、Llama 等。这会引入新的资源考量显存/内存占用这是最大的挑战。7B 参数的模型量化后可能需要 4-8GB 显存13B 模型则需要更多。务必根据你的硬件条件选择模型。推理速度本地推理速度通常慢于云端 API尤其是首次加载。需要优化推理库如 vLLM, llama.cpp的配置。启动方式你需要额外启动一个模型服务如 OpenAI 格式的兼容 API然后将PlannerAgent中的OpenAI客户端指向这个本地服务地址。8. 常见问题与排查方法在开发和运行 Agent 系统时你可能会遇到以下典型问题。这里提供排查思路。问题现象可能原因排查方式解决方案服务启动失败端口被占用端口 8000 已被其他程序如另一个开发服务使用。运行netstat -ano | findstr :8000(Win) 或lsof -i:8000(Mac/Linux) 查看占用进程。1. 终止占用进程。2. 修改启动命令中的端口号如--port 8001。调用/agent/query接口返回 500 内部错误1. API 密钥未设置或错误。2. Skill 执行过程中抛出异常。3. LLM 返回内容格式不符合预期。1. 检查终端或日志中的详细错误堆栈。2. 检查.env文件是否在项目根目录变量名是否正确。3. 打印plan_and_execute方法中的中间变量。1. 确认.env文件存在且密钥有效。2. 在 Skill 和 Tool 函数中加入更细致的异常捕获和日志。3. 增强 LLM 输出的格式校验和错误处理。Agent 选择了错误的 Skill1. 提示词System Prompt描述不清晰。2. 可用的 Skill 列表描述有歧义。3. 用户查询本身模糊。1. 打印出发送给 LLM 的完整提示词检查其清晰度。2. 测试不同的查询观察规律。1. 优化技能描述使其职责更分明。2. 在规划阶段引入“置信度”评分或让 LLM 返回多个候选技能并排序。3. 对于模糊查询设计一个澄清对话的流程。Skill 执行成功但结果不对1. 参数提取错误。2. 工具函数Tool本身的逻辑或依赖的第三方 API 有问题。1. 检查 LLM 提取出的参数 JSON 是否正确。2. 单独测试工具函数传入相同参数。1. 优化参数提取的提示词或使用更结构化的输出如 Pydantic 模型来约束 LLM。2. 修复工具函数的 bug 或检查第三方 API 状态。批量任务中部分请求失败1. 网络波动。2. 达到大模型 API 的速率限制。3. 个别查询触发了未知异常。1. 查看批量任务脚本中的错误日志。2. 检查第三方 API 控制台的用量和限流信息。1. 在批量处理器中为每个任务添加独立的重试机制。2. 在批量任务中增加请求间隔如time.sleep(0.5)。3. 确保每个任务都被try...except包裹。服务响应越来越慢1. 内存泄漏如未释放的大对象。2. 数据库或外部连接未正确关闭。3. 代码中存在同步阻塞操作。1. 使用内存 profiling 工具如memory_profiler。2. 检查代码中是否有全局列表或缓存无限增长。1. 审查代码确保资源正确释放。2. 将可能的阻塞 I/O 操作如文件读写、网络请求改为异步。3. 对于长时间运行的服务定期重启或使用进程管理工具如gunicorn配合多 worker。9. 最佳实践与使用建议基于以上实践我们总结出一些构建稳健、可维护 Agent 系统的最佳实践。技能Skill设计原则单一职责一个 Skill 只做一件事并把它做好。例如WeatherQuerySkill只负责查询天气不要让它同时去查航班。明确接口使用像 Pydantic 这样的库来严格定义 Skill 的输入和输出 Schema这有助于自动化验证和生成文档。可测试性每个 Skill 都应该可以脱离 Agent 框架进行独立单元测试。工具Tool管理统一注册中心维护一个全局的工具注册表Agent 可以通过名称动态查找和调用工具。这比硬编码在 Agent 类中更灵活。工具描述至关重要给每个工具编写清晰、无歧义的描述这是 LLM 能否正确使用它的关键。描述应包括功能、输入参数说明和输出示例。提示词Prompt工程模块化将 System Prompt 分解为角色定义、可用工具描述、输出格式要求等模块便于维护和组合。少样本Few-Shot学习在 Prompt 中提供几个正确调用工具的示例能显著提升 LLM 规划和使用工具的准确性。持续迭代将出错的案例收集起来分析是规划错误、参数提取错误还是工具本身错误并据此优化 Prompt。工作流Workflow编排可视化设计对于复杂工作流考虑使用像Prefect或Airflow这样的工作流引擎它们提供可视化界面和强大的调度、监控能力。状态持久化长时间运行或可能中断的工作流需要将执行状态保存到数据库以便恢复。错误处理与补偿在工作流中定义明确的错误处理策略例如重试、回滚或转人工。安全与合规输入输出过滤对用户输入和模型输出进行必要的过滤和审查防止注入攻击或生成不当内容。权限控制不同的 Skill/Tool 可能对应不同的数据或操作权限。在调用前进行权限校验。审计日志记录每一次 Agent 的决策过程、调用的工具和结果便于事后审计和问题追溯。从本教程构建的最小可行系统出发你可以向多个方向深入探索例如引入更强大的开源框架如 LangChain、LangGraph、Dify、Transformers Agents来获得更成熟的基础设施尝试更复杂的 Agent 架构如 ReAct、CoT 或 Multi-Agent 协作将系统部署到云服务器并为其开发一个聊天界面。这个过程的本质是让 AI 从“感知”走向“行动”而你已经掌握了让 AI 迈出第一步的关键拼图。