AI Agent开发实战：从工具调用到工作流编排的完整指南

这类教程最值得先看的不是它有多少集、标题有多吸引人，而是它到底能不能帮你把“AI Agent”这个听起来很玄乎的概念，变成能动手写、能跑起来、能解决实际问题的代码和项目。很多人学了一堆理论，看了一堆视频，最后还是不知道一个Agent从零到一该怎么搭，怎么调试，怎么处理失败重试和状态管理。这篇文章就围绕“Agent Skills”这个核心，拆解一个真正能跑通的开发流程，从环境准备、核心概念、代码实现到生产化思考，带你走一遍我实际落地时会关注的每个环节。

如果你刚接触AI应用开发，或者想从调用API进阶到设计自主工作流，这篇文章能帮你建立一个清晰的实操框架。如果你已经看过一些教程但感觉还是云里雾里，这里会重点讲那些教程里经常一笔带过，但实际开发中一定会遇到的“坑点”，比如工具调用格式、长上下文管理、任务拆解逻辑和错误处理。

1. 先拆解“Agent Skills”：它到底指什么，不是什么

看到“Agent Skills”这个词，很多人第一反应是“让AI Agent学会各种技能”。这个理解方向没错，但太宽泛，不利于动手。在实际的工程语境里，我们可以把它拆解成三个可操作、可实现的层面。

1.1 核心是“工具调用”与“工作流编排”

一个AI Agent的“Skill”，最基础的体现就是它能调用外部工具。这不仅仅是让大模型说一句“我去查一下天气”，而是要在代码层面实现：模型能生成结构化的工具调用请求（比如一个符合特定JSON Schema的指令），你的程序能解析这个请求，去执行真正的函数（调用天气API），再把执行结果以结构化的方式返回给模型，让模型基于结果进行下一步推理或回答。

所以，第一个要落地的“Skill”就是工具调用能力。这需要你：

1. 选择一个支持“Function Calling”或“Tool Calling”的大模型API（如OpenAI GPT-4, Claude, 国内的一些主流模型）。
2. 在你的代码里，明确定义工具（函数）的名称、描述、参数格式。
3. 在请求模型时，把这些工具定义传给模型。
4. 解析模型的返回，识别出它“想调用哪个工具”以及“参数是什么”。
5. 执行对应的本地函数或远程API调用。
6. 将调用结果再次放入对话历史，请求模型生成最终回答。

这个过程本身就是一个可复用的技能单元。多个这样的单元，按照一定的逻辑（顺序、分支、循环）组合起来，就是一个工作流（Workflow）。比如“先查天气，再根据天气推荐穿衣，最后生成一段出行提醒文案”，这就是一个由三个技能组成的工作流。

1.2 不等于“Prompt Engineering”的简单堆砌

有些教程会把Agent Skills等同于写更复杂的提示词（Prompt）。这有一定道理，因为提示词定义了Agent的角色和目标。但仅有提示词是不够的。一个健壮的Agent Skill必须包含错误处理、状态管理和数据验证。

例如，你让Agent调用一个获取股票价格的工具。如果工具返回“股票代码不存在”，你的Agent Skill不应该就此崩溃或胡言乱语，而应该能捕获这个错误，并可能触发重试（比如检查代码格式）、或向用户请求澄清、或执行一个备选方案（如查询相关公司列表）。这个“捕获错误-决策下一步”的逻辑，是Skill实现中比写提示词更关键的部分。

1.3 最终目标是“自主任务拆解与执行”

高阶的Agent Skills，是让Agent能够理解一个模糊的、复杂的人类指令（比如“帮我策划一个周末的北京出游计划”），然后自主将其拆解成一系列具体的、可执行的子任务（查天气、查景点开放时间、查交通路线、推荐餐厅、估算预算），并协调相应的工具按顺序或并行执行这些子任务，最后汇总成一个完整的答案。

这涉及到更复杂的能力：任务规划（Planning）、记忆（Memory）和多步推理（Reasoning）。目前，这通常通过“ReAct”（Reasoning + Acting）模式或利用具有强大规划能力的大模型（如GPT-4）来实现。对于初学者，我建议先不要追求这种完全自主的智能体，而是从实现一个预设好步骤的工作流开始。先让流程在你的控制下跑通，再逐步尝试让模型来决策步骤。

2. 动手前：明确你的开发环境与资源边界

在写第一行代码之前，先花十分钟确认你的环境。很多“跑不通”的问题都出在这里。

2.1 模型API：选型与成本控制

你需要一个能够处理工具调用的大模型服务。以下是常见的选项及考量：

我的建议：如果你是第一次开发Agent，为了减少环境干扰，优先使用一个你最容易获得、文档清晰的云API。先确保核心逻辑（调用-执行-返回）能通，再考虑换模型或本地化。同时，务必在账户里设置用量告警，避免意外高额账单。

2.2 编程语言与框架：Python是主流，但框架选择有讲究

Python是绝对的主流，生态最完善。你不必纠结于语言，但需要选一个开发框架来降低复杂度。

• 从零开始手写：仅使用requests库调用API，自己解析JSON。只推荐用于理解最底层原理，生产项目不推荐，因为会重复造轮子且易出错。
• 使用SDK + 自定义逻辑：使用OpenAI官方Python SDK等，它封装了工具调用的格式。你需要自己管理对话历史、工具注册和结果处理。这是很好的学习路径。
• 使用Agent开发框架：
  • LangChain / LangGraph：生态庞大，组件丰富，但抽象层次高，初学者容易陷入框架细节而看不清本质。适合快速搭建复杂原型。
  • LlamaIndex：更专注于数据查询和RAG（检索增强生成），与Agent结合紧密。
  • Semantic Kernel(微软)：跨语言，设计理念不错，但Python版生态相对较新。
  • AutoGen(微软)：专注于多智能体对话和协作，适合研究型或复杂协作场景。

给新手的路线图：

1. 第一周：用OpenAI SDK（或你选的模型SDK）纯手写一个最简单的工具调用Demo。目的是彻底搞懂“请求-响应-执行-再请求”这个循环。
2. 第二周：尝试用LangChain重构同一个Demo。感受框架帮你做了什么（如自动管理聊天历史、简化工具定义），同时理解它带来的复杂性。
3. 第三周及以后：根据你的项目需求（是否需要复杂工作流、多智能体、强大RAG）选择深耕一个框架，或基于SDK打造自己的轻量级框架。

2.3 关键依赖清单

一个干净的Python虚拟环境是必须的。以下是核心依赖示例（以OpenAI为例）：

# 创建虚拟环境 python -m venv venv_agent # 激活 (Windows) venv_agent\Scripts\activate # 激活 (macOS/Linux) source venv_agent/bin/activate # 安装核心依赖 pip install openai python-dotenv # 可选：如果你打算用LangChain pip install langchain langchain-openai

将你的API Key放在项目根目录的.env文件中，使用python-dotenv加载，永远不要硬编码在代码里。

# .env 文件内容 OPENAI_API_KEY=sk-your-key-here

3. 从零实现第一个核心Skill：天气查询Agent

现在，我们抛开所有框架，用最直接的方式实现一个能查询天气的Agent。这个过程会暴露所有关键环节。

3.1 第一步：定义“工具”（Tool）

工具本质上是一个函数，以及描述这个函数的“说明书”（Schema）。我们先写函数：

import requests import json from typing import Optional def get_current_weather(location: str, unit: str = "celsius") -> str: """ 获取指定城市的当前天气情况。 Args: location: 城市名，例如“北京”、“San Francisco”。 unit: 温度单位，“celsius” 或 “fahrenheit”。默认为“celsius”。 Returns: 描述天气的字符串。 """ # 警告：这是一个模拟函数。真实场景应调用如OpenWeatherMap等API。 # 这里为了演示，返回模拟数据。 weather_info = { "location": location, "temperature": "22", "unit": unit, "forecast": ["sunny", "windy"], "humidity": "65%" } return json.dumps(weather_info)

接下来，创建工具的“说明书”。这个说明书要符合模型能理解的格式（OpenAI格式）：

tools = [ { "type": "function", "function": { "name": "get_current_weather", "description": "获取某个城市的当前天气", "parameters": { "type": "object", "properties": { "location": { "type": "string", "description": "城市或地区名，例如：北京，东京", }, "unit": { "type": "string", "enum": ["celsius", "fahrenheit"], "description": "温度单位", } }, "required": ["location"], }, }, } ]

关键点：description字段至关重要，模型靠它来决定是否以及如何调用这个工具。要写得清晰、具体。

3.2 第二步：发起对话，让模型决定是否调用工具

我们使用OpenAI SDK（openai库）来发起聊天请求，并把工具定义传给它。

import os from openai import OpenAI from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 client = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) def run_conversation(user_query: str): # 步骤1: 将用户消息和工具定义发送给模型 response = client.chat.completions.create( model="gpt-3.5-turbo", # 或 "gpt-4" messages=[{"role": "user", "content": user_query}], tools=tools, # 关键：传入工具定义 tool_choice="auto", # 让模型自动决定是否调用工具 ) response_message = response.choices[0].message tool_calls = response_message.tool_calls # 检查模型是否想调用工具 # 步骤2: 如果模型想调用工具，我们就能看到具体的调用请求 if tool_calls: print(f"模型决定调用工具。调用请求: {tool_calls}") # 通常只有一个tool_call，我们遍历处理 for tool_call in tool_calls: function_name = tool_call.function.name function_args = json.loads(tool_call.function.arguments) # 步骤3: 找到本地对应的函数并执行 available_functions = { "get_current_weather": get_current_weather, } function_to_call = available_functions[function_name] function_response = function_to_call(**function_args) print(f"工具 `{function_name}` 执行结果: {function_response}") # 步骤4: 将工具执行结果作为新的消息，再次发送给模型 # 这一步是关键，让模型基于结果生成最终回答 second_response = client.chat.completions.create( model="gpt-3.5-turbo", messages=[ {"role": "user", "content": user_query}, response_message, # 模型第一次的回复（包含工具调用意图） { "role": "tool", "content": function_response, "tool_call_id": tool_call.id, # 必须关联对应的tool_call }, ], ) return second_response.choices[0].message.content else: # 模型没有调用工具，直接返回回答 return response_message.content # 测试 if __name__ == "__main__": query = "北京今天天气怎么样？" final_answer = run_conversation(query) print(f"用户: {query}") print(f"Agent: {final_answer}")

运行这段代码，你会看到类似以下的输出：

模型决定调用工具。调用请求: [ChatCompletionMessageToolCall(id='call_abc123', function=Function(arguments='{"location": "北京", "unit": "celsius"}', name='get_current_weather'), type='function')] 工具 `get_current_weather` 执行结果: {"location": "北京", "temperature": "22", "unit": "celsius", "forecast": ["sunny", "windy"], "humidity": "65%"} 用户: 北京今天天气怎么样？ Agent: 北京目前天气晴朗，有风，气温22摄氏度，湿度65%。

3.3 第三步：复盘这个简单流程里的“坑点”

这个Demo虽然简单，但包含了Agent Skill最核心的循环。在这个过程中，最容易出问题的地方是：

1. 工具定义（Schema）不匹配：你的函数参数名、类型、是否必须，必须和tools里定义的parameters完全一致。模型生成的参数会严格按照这个Schema来。
2. 忘记传递tool_call_id：在第二次请求，将工具执行结果返回给模型时，tool_call_id必须对应第一次请求中的那个调用ID。这是模型区分多个并行工具调用的关键。
3. 模型不调用工具：如果模型没有调用工具，可能因为：
   • 工具描述不清：description没写明白这个工具是干什么的。
   • 用户问题太模糊：比如“今天怎么样？”，模型可能直接回答而不调用天气工具。
   • 模型能力或温度（temperature）参数影响：可以尝试更清晰的用户指令，或使用能力更强的模型（如从gpt-3.5-turbo切换到gpt-4）。
4. 网络或API错误：工具函数里调用真实API时，必须有超时和重试机制，并且要把错误信息妥善地返回给模型（例如“网络请求失败，请稍后再试”），而不是让整个程序崩溃。

4. 从单技能到多技能与工作流编排

单个技能跑通后，接下来要解决两个问题：如何管理多个技能？如何让它们按顺序或条件执行？

4.1 管理多个工具：注册与路由

当你有多个工具时（如get_weather,search_web,send_email），你需要一个中心化的地方来注册和管理它们。一个简单的字典仍然有效，但更好的做法是创建一个ToolRegistry类：

class ToolRegistry: def __init__(self): self._tools = {} self._schemas = [] def register(self, name: str, func: callable, schema: dict): self._tools[name] = func self._schemas.append(schema) def get_tool(self, name: str) -> callable: return self._tools.get(name) def get_schemas(self) -> list: return self._schemas # 使用示例 registry = ToolRegistry() registry.register("get_current_weather", get_current_weather, tools[0]) # 假设tools[0]是天气工具的schema # 在run_conversation函数中，从registry获取schemas和具体函数 available_functions = registry._tools tools_for_api = registry.get_schemas()

这样，新增工具只需要调用register一次，代码更清晰。

4.2 实现顺序工作流：手动编排 vs 模型规划

方案A：手动编排（确定性强，推荐新手）你作为开发者，预先定义好步骤。例如，“先查天气，再根据天气推荐活动”。

def plan_weekend(activity: str): # 步骤1: 获取天气 weather = get_current_weather("北京") weather_data = json.loads(weather) # 步骤2: 根据天气和活动类型，生成推荐（这里可以再调用一次LLM） recommendation_prompt = f""" 天气信息：{weather_data}。 用户想进行的活动类型：{activity}。 请结合天气，给出是否适合进行此活动的建议，并说明理由。 """ # 调用LLM生成推荐... return final_recommendation

这种方式完全由你控制流程，适合业务逻辑固定的场景。

方案B：让模型规划（更灵活，但不可控）你可以设计一个“规划师”Agent，它的工具就是调用其他技能。或者，直接利用支持复杂推理的模型（如GPT-4），通过一个包含所有工具定义的提示词，让它自己决定调用顺序。这更接近“智能”体，但也更难以调试和保证稳定性。

我的建议：在生产的初期，优先采用手动编排。把复杂任务拆解成几个明确的步骤，每个步骤可以是一个Agent Skill。这样每个环节都可测试、可监控。等核心流程稳定后，再尝试将部分决策权交给模型。

4.3 引入状态管理：记忆（Memory）

一个有用的Agent通常需要记住之前的对话。最简单的记忆就是保存整个对话历史（messages列表）。但长对话下，这会导致token消耗巨大且可能超出模型上下文限制。

更高级的记忆管理包括：

• 摘要记忆：定期将长对话总结成摘要，只保留摘要和最近几条消息。
• 向量存储记忆：将历史对话片段转换成向量存入数据库（如Chroma, Pinecone），需要时进行语义检索。
• 结构化记忆：将关键信息（如用户偏好、任务状态）存入结构化的数据库。

对于新手，先从维护一个全局的messages列表开始，每次对话都将其传入。同时，密切关注token使用量，当对话轮次增多时，考虑实现一个简单的“滑动窗口”记忆，只保留最近N轮对话。

5. 向生产环境迈进：稳定性、监控与部署思考

能让一个Agent在笔记本上跑起来，和能让它稳定服务用户，是两回事。以下是走向生产必须考虑的几点。

5.1 健壮性设计：错误处理与重试

• 工具调用失败：网络超时、API限流、参数错误。你的代码必须捕获这些异常，并决定是重试、跳过还是向用户报错。
• 模型响应异常：模型可能返回无法解析的JSON，或调用了未定义的工具。需要设计一个fallback机制。
• 超时控制：给整个Agent对话流程设置总超时，防止死循环。

import tenacity from openai import APITimeoutError @tenacity.retry( stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(multiplier=1, min=4, max=10), retry=tenacity.retry_if_exception_type((APITimeoutError, requests.exceptions.Timeout)), ) def call_llm_with_retry(client, messages, tools): """一个带重试的LLM调用函数示例""" try: response = client.chat.completions.create( model="gpt-3.5-turbo", messages=messages, tools=tools, timeout=30.0, # 设置单次请求超时 ) return response except APITimeoutError: # 记录日志 raise # 让tenacity捕获并重试

5.2 可观测性：日志与监控

在生产中，你必须知道你的Agent在干什么。

• 结构化日志：记录每个关键步骤（用户输入、模型请求、工具调用、工具结果、最终输出）。使用logging模块，并输出JSON格式，方便后续收集分析。
• 关键指标：记录每次调用的耗时、token消耗、工具调用次数、失败率。
• 链路追踪（Trace）：为每个用户会话生成一个唯一ID，将这个ID贯穿所有日志和调用，这样当出现问题时，可以完整复现该用户的处理流程。

5.3 部署模式：API服务 vs 异步任务

• 同步API服务：使用FastAPI、Flask等框架，将Agent封装成一个HTTP端点。适合实时交互场景。注意设置合理的请求超时和并发限制。
• 异步任务队列：对于耗时较长的复杂工作流（如处理一个文档、生成一份报告），更适合放入Celery、Dramatiq或RQ这样的任务队列中异步执行，通过WebSocket或轮询通知用户结果。

一个简单的FastAPI部署示例：

from fastapi import FastAPI, HTTPException from pydantic import BaseModel import logging app = FastAPI() logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class AgentRequest(BaseModel): query: str session_id: str = None @app.post("/chat") async def chat_with_agent(request: AgentRequest): logger.info(f"Session {request.session_id}: Received query - {request.query}") try: # 这里调用你之前写好的 run_conversation 逻辑 answer = run_conversation(request.query) logger.info(f"Session {request.session_id}: Successfully generated answer") return {"answer": answer, "session_id": request.session_id} except Exception as e: logger.error(f"Session {request.session_id}: Error - {str(e)}", exc_info=True) raise HTTPException(status_code=500, detail="Agent processing failed")

5.4 成本与性能优化

• 缓存：对频繁且结果不变的工具调用（如某些查询）结果进行缓存。
• 选择合适模型：在非核心推理步骤使用更便宜、更快的模型（如gpt-3.5-turbo），只在需要复杂规划或创意生成时使用强大模型（如GPT-4）。
• 精简上下文：定期清理对话历史中的无关信息，使用摘要，避免token无意义增长。

Agent Skills的学习和实践是一个从“点”（单个工具调用）到“线”（工作流）再到“体”（多智能体协作、复杂记忆与规划）的过程。我建议你不要一开始就追求大而全的框架或复杂的多智能体设计。从今天这个最简单的天气查询Demo开始，亲手敲一遍代码，理解数据是如何在用户、模型和你写的函数之间流转的。把这个基础循环搞扎实了，后面无论遇到什么框架或高级概念，你都能一眼看穿它的本质。在实际项目中，先把一个用手动编排的、有完整错误处理的、日志清晰的核心工作流跑稳，这比一个看似智能但动不动就崩溃的“黑盒”要可靠得多。