企业官网建设流程全解析

1. 项目概述：这不是又一个“AI Agent框架”，而是一套可插拔的技能积木

你有没有遇到过这样的场景：花三天时间写了个能调用天气API的Agent，结果第二天产品经理说“用户还要查航班”，你只好再啃一遍航空公司的文档，重写一套认证、解析、错误重试逻辑；或者更糟——把天气和航班代码硬塞进同一个run()函数里，最后变成一团谁也不敢动的意大利面条？我干过这种事，而且不止一次。pydantic-ai-skills这个项目标题里的关键词——“Composable Agent Skills”（可组合的Agent技能）——不是营销话术，它直指当前AI应用开发中最痛的痛点：技能复用率低、耦合度高、调试成本爆炸。它不属于某个大而全的Agent平台，而是为Pydantic AI生态量身定制的一套“技能标准件”。你可以把它理解成乐高积木里的“齿轮模块”或“万向轮底座”：单个积木本身不构成完整机器人，但一旦你有了标准化的齿轮齿距和轴孔尺寸，就能把不同厂商、不同功能的模块无缝拼在一起。这个项目的核心价值，不在于它自己能做什么，而在于它让“写一个新技能”这件事，从一场需要重新设计底盘、电机、传感器的工程攻坚，降维成一次拧紧几颗螺丝的机械装配。它面向的不是刚学Python的大学生，而是已经用Pydantic AI搭过至少两个真实Agent、正被重复造轮子折磨得想删库跑路的中高级开发者。如果你还在为每个新API都得重写一遍validate_response()、handle_rate_limit()、format_for_llm()而烦躁，那这篇拆解就是为你写的。

2. 内容整体设计与思路拆解：为什么是“技能”而不是“Agent”？

2.1 核心范式转移：从“构建完整Agent”到“组装标准化技能”

在深入代码前，必须先厘清一个根本性设计哲学：pydantic-ai-skills 的目标不是替代你现有的Agent框架，而是成为它的“技能供应商”。这背后有三层深意。第一层是工程现实——绝大多数业务场景中的Agent，其核心复杂度并不在决策链路（LLM调用、工具选择、记忆管理），而在于与外部世界交互的“脏活累活”：处理千奇百怪的API响应格式、应对瞬息万变的认证方式、兜底网络抖动导致的超时、将非结构化文本清洗成结构化数据。这些工作占了实际开发时间的70%以上，却极少被抽象复用。第二层是生态协同——Pydantic AI本身是一个强调类型安全、数据验证、序列化的轻量级生态，它天然排斥那种把所有功能打包进一个巨型类的设计。如果强行把航班查询、股票行情、数据库查询都塞进一个MySuperAgent类里，就等于用Pydantic的“类型之矛”去攻自己“模块化之盾”，自相矛盾。第三层是演进韧性——当航空公司升级了他们的OpenAPI规范，你只需要更新FlightSearchSkill这一个模块，而不用牵一发而动全身地重构整个Agent。我去年维护一个金融分析Agent时，就因为某家券商突然废弃了旧版REST API，被迫在48小时内紧急回滚并重写所有数据获取逻辑，那次教训让我彻底放弃了“大一统Agent”的幻想。pydantic-ai-skills 的设计者显然也踩过同样的坑，所以他们选择了一条更“笨”但更可持续的路：把技能定义成独立、自治、可测试的单元，每个单元只做一件事，并且这件事必须做得足够干净、足够健壮。

2.2 “Composable”背后的三重技术契约

“可组合”这个词听起来很虚，但在pydantic-ai-skills里，它被具象化为三条硬性技术契约，任何想接入这个生态的技能都必须遵守：

1. 输入契约：强制使用Pydantic v2模型定义参数
   你不能用一个裸字典{"origin": "PEK", "dest": "SHA"}来接收航班查询请求。你必须定义一个继承自BaseModel的类，比如：

   from pydantic import BaseModel, Field class FlightSearchInput(BaseModel): origin: str = Field(..., min_length=3, max_length=3, pattern=r'^[A-Z]{3}$') destination: str = Field(..., min_length=3, max_length=3, pattern=r'^[A-Z]{3}$') date: date = Field(..., description="Flight date in YYYY-MM-DD format")

   这个看似繁琐的步骤，实则解决了三个致命问题：一是自动完成参数校验（机场三字码必须是大写英文字母），二是生成清晰的OpenAPI文档供LLM理解，三是为后续的类型提示、IDE自动补全提供基础。我试过直接传字典，结果LLM在调用时把"date": "2024-10-01"错传成"date": "Oct 1st"，整个调用链直接崩掉，而Pydantic模型会在进入技能前就抛出ValidationError，把错误拦截在最外层。

2. 输出契约：统一返回SkillResult泛型包装器
   所有技能的返回值，无论内部逻辑多么复杂，最终都必须包装成SkillResult[YourOutputModel]。这个泛型包装器不是画蛇添足，它内置了success: bool、error: Optional[str]、data: Optional[T]三个字段。这意味着，当你在Agent的主流程里调用result = flight_skill.run(input)时，你永远不需要猜result是字典、是字符串、还是None。你只需检查if result.success:，然后安全地使用result.data。这个设计直接消灭了90%的AttributeError: 'NoneType' object has no attribute 'flights'这类低级但高频的崩溃。我在一个电商比价Agent里，曾因某个价格API返回空数组而没做判空，导致后续所有计算都基于None进行，花了整整一个下午才定位到源头。

3. 生命周期契约：显式声明依赖与资源管理
   技能不是孤立运行的。它可能需要一个HTTP会话、一个数据库连接池、一个缓存客户端。pydantic-ai-skills要求你通过__init__方法的参数签名，明确声明这些依赖。例如：

   class FlightSearchSkill(Skill[FlightSearchInput, FlightSearchOutput]): def __init__(self, http_client: AsyncClient, cache: RedisCache): self.http_client = http_client self.cache = cache

   这个设计强迫你在设计阶段就思考“这个技能到底需要什么”，而不是等到上线后才发现并发量上来时，每个请求都新建一个HTTP会话，把服务器的文件描述符耗尽。更重要的是，它为依赖注入（DI）框架提供了清晰的接口，你可以轻松地用pytest的monkeypatch替换http_client来写单元测试，而不用启动一个真实的Mock服务。

2.3 为什么选择Pydantic AI而非LangChain或LlamaIndex？

这个问题常被问起，答案很务实：不是技术优劣，而是场景匹配度。LangChain是一个庞大的、企业级的Agent操作系统，它自带记忆、路由、回调、追踪等一整套基础设施。但如果你的团队只有3个人，要在一个季度内上线5个垂直领域的小型Agent（比如HR政策问答、IT工单助手、销售合同初审），LangChain的厚重感反而成了负担。它的配置项多如牛毛，一个简单的Tool定义可能就要嵌套三四层类。而Pydantic AI的哲学是“最小可行抽象”——它只提供Agent、Tool、Message这几个最核心的基类，其余全部交由开发者用Pydantic模型去定义。这就像给你一把瑞士军刀，而不是一台数控机床。pydantic-ai-skills正是在这种轻量哲学下生长出来的“配件”。它不关心你的Agent用什么LLM、用什么记忆策略，它只关心：“你这个技能，输入是什么？输出是什么？怎么执行？”这种松耦合，让它能无缝集成到任何基于Pydantic构建的Agent中，无论是你自己手写的极简版，还是社区里流行的pydantic-ai-agent库。我做过一个对比实验：用同样的航班查询逻辑，在LangChain里实现一个StructuredTool，代码量是127行；在pydantic-ai-skills里，核心逻辑只有43行，剩下的都是类型定义和错误处理——而这43行，是真正和业务相关的代码。

3. 核心细节解析与实操要点：从零开始构建一个航班查询技能

3.1 环境准备与依赖安装：避开版本地狱

在动手写代码前，环境配置是第一个也是最容易翻车的环节。pydantic-ai-skills对Pydantic版本有严格要求，它深度依赖v2的@model_validator、@field_validator等新特性，而很多老项目还卡在v1.x。因此，强烈建议创建一个全新的虚拟环境，不要试图在现有项目里“凑合着用”。以下是经过我实测的、最稳妥的安装命令：

# 创建并激活新环境 python -m venv ./pydantic-skills-env source ./pydantic-skills-env/bin/activate # Linux/Mac # ./pydantic-skills-env/Scripts/activate # Windows # 安装核心依赖（注意顺序！） pip install "pydantic>=2.5.0,<3.0.0" # 必须锁定在2.x，3.0+有重大breaking change pip install "httpx>=0.25.0" # pydantic-ai-skills默认HTTP客户端 pip install "redis>=4.6.0" # 如果要用Redis缓存 pip install "pydantic-ai-skills>=0.3.0" # 当前最新稳定版

提示：不要用pip install pydantic-ai-skills而不加版本号。我曾因未指定版本，pip自动安装了预发布的0.4.0a1版本，结果发现其SkillResult的泛型约束在Python 3.9上存在类型推导bug，导致整个IDE的类型提示全部失效，白白浪费了半天时间排查。官方文档里提到的“推荐版本”往往滞后于实际发布，最可靠的方式是去PyPI页面查看Requires: Python >=3.9和Requires-Dist字段，然后手动指定一个已知稳定的版本。

3.2 技能骨架搭建：从Skill基类开始

一切始于继承Skill这个抽象基类。它不是一个装饰器，也不是一个函数，而是一个需要你认真填写的“技能蓝图”。让我们以航班查询为例，一步步构建：

from pydantic_ai_skills import Skill from pydantic import BaseModel, Field, field_validator from datetime import date from typing import List, Optional # 1. 定义输入模型（Input Contract） class FlightSearchInput(BaseModel): origin: str = Field( ..., description="Three-letter IATA airport code for departure (e.g., PEK)", min_length=3, max_length=3, pattern=r'^[A-Z]{3}$' ) destination: str = Field( ..., description="Three-letter IATA airport code for arrival (e.g., SHA)", min_length=3, max_length=3, pattern=r'^[A-Z]{3}$' ) date: date = Field(..., description="Flight date in YYYY-MM-DD format") @field_validator('date') @classmethod def date_must_be_future(cls, v: date) -> date: """确保日期不早于今天""" from datetime import datetime today = datetime.now().date() if v < today: raise ValueError('Flight date must be today or in the future') return v # 2. 定义输出模型（Output Contract） class FlightInfo(BaseModel): flight_number: str = Field(..., description="e.g., CA123") departure_time: str = Field(..., description="HH:MM in local time") arrival_time: str = Field(..., description="HH:MM in local time") duration_minutes: int = Field(..., ge=30, le=1200, description="Flight duration in minutes") price_cny: float = Field(..., ge=0.0, description="Price in Chinese Yuan") class FlightSearchOutput(BaseModel): flights: List[FlightInfo] = Field(..., min_items=1, description="List of available flights") currency: str = Field(default="CNY", description="Currency of the prices") # 3. 继承Skill基类，声明泛型参数 class FlightSearchSkill(Skill[FlightSearchInput, FlightSearchOutput]): """ A composable skill to search for flights between two airports on a given date. This skill demonstrates input validation, external API integration, and error handling. """ # 技能元信息（用于Agent发现和LLM理解） name: str = "flight_search" description: str = ( "Search for available flights between two airports on a specific date. " "Returns flight numbers, times, durations, and prices." )

这段代码里藏着几个关键细节。首先是@field_validator的使用——它不只是校验格式，更是向LLM传递业务规则的“人话说明书”。当你把FlightSearchInput模型喂给LLM时，它不仅能看见origin: str，还能看到description="Three-letter IATA airport code..."和@field_validator里的注释。这极大提升了LLM生成正确参数的概率。其次是FlightInfo模型里的ge=30, le=1200约束，这不仅是防御性编程，更是对LLM的一种“温柔提醒”：如果它生成了一个duration_minutes=5的假数据，Pydantic会在SkillResult包装前就报错，而不是让错误数据流入下游系统。最后是name和description字段，它们不是可选的。pydantic-ai-skills的Agent调度器会扫描所有注册的技能，根据name去匹配LLM的工具调用请求，根据description去生成Prompt中的工具列表。我曾把name写成"search_flights"，而LLM的调用是{"name": "flight_search"}，结果技能永远无法被触发，调试了两个小时才发现是命名不一致。

3.3 核心执行逻辑：run方法的编写艺术

run方法是技能的“心脏”，它必须是异步的（async def），因为几乎所有外部API调用都是I/O密集型的。这里没有魔法，只有扎实的工程实践：

import httpx from pydantic_ai_skills import SkillResult class FlightSearchSkill(Skill[FlightSearchInput, FlightSearchOutput]): # ... 上面的定义保持不变 ... def __init__(self, http_client: httpx.AsyncClient): self.http_client = http_client async def run(self, input: FlightSearchInput) -> SkillResult[FlightSearchOutput]: """ Main execution logic. This is where the real work happens. Returns a SkillResult wrapping either success data or an error. """ try: # Step 1: 构建API请求URL和参数 # 注意：这里使用的是模拟的API端点，实际项目中请替换为真实服务商 url = "https://api.example-airlines.com/v1/flights/search" params = { "origin": input.origin.upper(), # 确保大写，符合IATA标准 "destination": input.destination.upper(), "date": input.date.isoformat() # 转换为YYYY-MM-DD字符串 } # Step 2: 发起异步HTTP请求，设置合理的超时 # 关键：不要用默认超时！网络抖动是常态 response = await self.http_client.get( url, params=params, timeout=httpx.Timeout(15.0, connect=5.0) # 总超时15秒，连接超时5秒 ) # Step 3: 检查HTTP状态码 response.raise_for_status() # 自动抛出HTTPStatusError # Step 4: 解析JSON响应 raw_data = response.json() # Step 5: 将原始JSON映射到我们的Pydantic输出模型 # 这是最重要的一步：数据清洗和转换 output_data = FlightSearchOutput( flights=[ FlightInfo( flight_number=flight["flightNumber"], departure_time=flight["departureTime"], arrival_time=flight["arrivalTime"], duration_minutes=int(flight["durationMinutes"]), price_cny=float(flight["priceCNY"]) ) for flight in raw_data.get("flights", []) ], currency=raw_data.get("currency", "CNY") ) # Step 6: 返回成功结果 return SkillResult.success(output_data) except httpx.TimeoutException as e: # 网络超时是最高频的错误，必须单独捕获并友好提示 error_msg = f"Flight search timed out after {e.request.timeout} seconds. Please try again later." return SkillResult.failure(error_msg) except httpx.HTTPStatusError as e: # 处理4xx/5xx错误 if e.response.status_code == 400: error_msg = "Invalid request parameters. Please check airport codes and date." elif e.response.status_code == 401: error_msg = "Authentication failed. Please contact system administrator." elif e.response.status_code == 429: error_msg = "Too many requests. Please wait a minute and try again." else: error_msg = f"Server error: {e.response.status_code}. Please try again later." return SkillResult.failure(error_msg) except KeyError as e: # 响应JSON结构与预期不符，这是API变更的典型信号 error_msg = f"Unexpected API response format. Missing key: {e}. The airline API may have changed." return SkillResult.failure(error_msg) except Exception as e: # 捕获所有其他未预期异常，作为最后的保险 error_msg = f"An unexpected error occurred: {str(e)}" return SkillResult.failure(error_msg)

这段run方法的编写，体现了三个核心原则。第一是防御性编程：每一个外部依赖点（HTTP请求、JSON解析、字段访问）都包裹在try...except里，并且针对不同错误类型给出不同的、对用户友好的提示。第二是数据主权：我们绝不信任外部API返回的原始JSON。raw_data.get("flights", [])确保了即使API返回空数组，也不会导致KeyError；int()和float()的强制转换，是为了把字符串数字转成我们需要的类型，同时int()会自动处理"120"和120两种情况。第三是错误分类：TimeoutException、HTTPStatusError、KeyError被分门别类地处理，而不是一股脑地丢给一个except Exception。这让你在日志里一眼就能看出，是网络问题、服务端问题，还是数据格式问题。我曾经在一个生产环境中，因为没区分HTTPStatusError和KeyError，导致所有API变更都被记录为“未知错误”，运维同学花了两天时间才从海量日志里揪出是上游API把"price"字段改成了"price_cny"。

3.4 高级技巧：缓存、重试与日志注入

一个生产级的技能，绝不能只满足于“能跑通”。它还需要缓存、重试和可观测性。pydantic-ai-skills本身不提供这些，但它为你预留了完美的扩展点：

import asyncio import logging from redis import Redis from tenacity import retry, stop_after_attempt, wait_exponential # 在Skill类内部添加缓存和重试支持 class FlightSearchSkill(Skill[FlightSearchInput, FlightSearchOutput]): # ... 其他定义 ... def __init__( self, http_client: httpx.AsyncClient, cache: Optional[Redis] = None, logger: Optional[logging.Logger] = None ): self.http_client = http_client self.cache = cache self.logger = logger or logging.getLogger(__name__) @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10), reraise=True ) async def _fetch_from_api(self, url: str, params: dict) -> dict: """带指数退避重试的API调用""" response = await self.http_client.get(url, params=params, timeout=15.0) response.raise_for_status() return response.json() async def run(self, input: FlightSearchInput) -> SkillResult[FlightSearchOutput]: # Step 0: 生成缓存键（基于输入参数的确定性哈希） cache_key = f"flight:{input.origin}:{input.destination}:{input.date}" # Step 1: 尝试从缓存读取 if self.cache: try: cached_json = self.cache.get(cache_key) if cached_json: self.logger.info(f"Cache hit for {cache_key}") raw_data = json.loads(cached_json) # 直接跳过API调用，走数据映射流程 output_data = self._map_to_output(raw_data) return SkillResult.success(output_data) except Exception as e: self.logger.warning(f"Cache read failed: {e}") # Step 2: 执行带重试的API调用 try: raw_data = await self._fetch_from_api( "https://api.example-airlines.com/v1/flights/search", {"origin": input.origin, "destination": input.destination, "date": input.date.isoformat()} ) # Step 3: 映射数据（同上） output_data = self._map_to_output(raw_data) # Step 4: 写入缓存（TTL设为2小时，航班信息变化相对缓慢） if self.cache: try: self.cache.setex(cache_key, 7200, json.dumps(raw_data)) self.logger.info(f"Cache set for {cache_key}") except Exception as e: self.logger.warning(f"Cache write failed: {e}") return SkillResult.success(output_data) except Exception as e: # 重试失败后的兜底逻辑 return SkillResult.failure(f"Failed after 3 retries: {str(e)}") def _map_to_output(self, raw_data: dict) -> FlightSearchOutput: """提取数据映射逻辑，便于单元测试""" return FlightSearchOutput( flights=[ FlightInfo( flight_number=flight["flightNumber"], departure_time=flight["departureTime"], arrival_time=flight["arrivalTime"], duration_minutes=int(flight["durationMinutes"]), price_cny=float(flight["priceCNY"]) ) for flight in raw_data.get("flights", []) ], currency=raw_data.get("currency", "CNY") )

这个增强版展示了如何在不破坏Skill契约的前提下，优雅地注入企业级能力。@retry装饰器来自tenacity库，它比手写while循环重试更可靠、更易配置。缓存键的生成使用了f-string，简单直接，避免了引入复杂的哈希库。最关键的是_map_to_output这个私有方法——它把数据映射逻辑抽离出来，使得你可以为它单独写单元测试，而不用每次都mock HTTP请求。我在一个金融项目里，就因为把数据映射和HTTP调用混在一起，导致单元测试覆盖率始终上不去，后来重构后，测试速度提升了5倍，覆盖率从62%升到了94%。

4. 实操过程与核心环节实现：将技能集成到Agent工作流

4.1 技能注册与发现：让Agent“认识”你的技能

写好一个技能，只是完成了50%。另一半工作，是让Agent知道它的存在。pydantic-ai-skills采用了一种非常Pythonic的“自动发现”机制，它基于Python的importlib和pkgutil，扫描指定模块路径下的所有Skill子类。假设你的技能文件放在my_project/skills/flight.py，那么注册过程如下：

# my_project/agent.py from pydantic_ai_skills import SkillRegistry from my_project.skills.flight import FlightSearchSkill from my_project.skills.weather import WeatherForecastSkill from my_project.skills.database import SqlQuerySkill # 方式1：手动注册（适合少量技能，调试友好） registry = SkillRegistry() registry.register(FlightSearchSkill(http_client=your_http_client)) registry.register(WeatherForecastSkill(api_key="your-key")) registry.register(SqlQuerySkill(db_pool=your_db_pool)) # 方式2：自动扫描（适合大型项目，技能数量多） # 这会递归扫描my_project.skills包下的所有.py文件 registry = SkillRegistry.from_package("my_project.skills") # 方式3：混合模式（推荐！） registry = SkillRegistry() # 先自动扫描所有技能 registry.scan_package("my_project.skills") # 再手动覆盖或补充特定实例（比如需要传入不同的HTTP客户端） registry.register(FlightSearchSkill(http_client=fast_http_client))

注意：SkillRegistry本身不存储技能的实例，它只存储技能的“构造器”（即类本身）和初始化所需的参数签名。当你调用registry.get_skill("flight_search")时，它才会根据你之前注册时提供的参数，动态地instantiate一个新实例。这种设计保证了每个技能实例都是独立的、无状态的，避免了多线程/协程环境下的共享状态污染。我曾在一个高并发场景下，因为错误地在SkillRegistry里注册了一个全局的、带状态的http_client实例，导致不同用户的请求互相干扰，出现了极其诡异的“张三的航班查询返回了李四的天气预报”这种问题。

4.2 Agent主流程集成：如何让LLM“调用”你的技能

Agent的主流程，本质上是一个“规划-执行-反思”的循环。pydantic-ai-skills的集成点，就在“执行”这一步。以下是一个极简但完整的Agent示例，它展示了如何将SkillRegistry与一个LLM（这里用openai.ChatCompletion模拟）结合：

import asyncio import json from openai import AsyncOpenAI from pydantic_ai_skills import SkillRegistry, SkillResult class SimpleAgent: def __init__(self, registry: SkillRegistry, llm_client: AsyncOpenAI): self.registry = registry self.llm_client = llm_client async def run(self, user_query: str) -> str: # Step 1: 构建System Prompt，告诉LLM有哪些可用技能 system_prompt = f""" You are a helpful AI assistant. You can use the following tools to answer the user's question. Each tool has a name and a description. Choose the most appropriate tool and call it with the correct arguments. Available tools: {self._build_tools_description()} """ # Step 2: LLM生成工具调用请求（Function Calling） response = await self.llm_client.chat.completions.create( model="gpt-4-turbo", messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_query} ], tools=self._build_tools_schema(), # 生成OpenAPI格式的tools schema tool_choice="auto" ) # Step 3: 解析LLM的tool_calls tool_calls = response.choices[0].message.tool_calls if not tool_calls: return response.choices[0].message.content # Step 4: 对每个tool_call，执行对应的技能 results = [] for tool_call in tool_calls: try: # 从registry中获取技能实例 skill = self.registry.get_skill(tool_call.function.name) if not skill: raise ValueError(f"Unknown skill: {tool_call.function.name}") # 解析LLM传入的参数（JSON字符串） args = json.loads(tool_call.function.arguments) # 使用Pydantic模型验证并解析参数 input_model = skill.input_model parsed_input = input_model(**args) # 执行技能 result = await skill.run(parsed_input) # 将SkillResult转换为LLM可理解的格式 if result.success: tool_result = { "name": tool_call.function.name, "content": json.dumps({"success": True, "data": result.data.model_dump()}) } else: tool_result = { "name": tool_call.function.name, "content": json.dumps({"success": False, "error": result.error}) } results.append(tool_result) except Exception as e: # 技能执行失败，返回错误信息给LLM results.append({ "name": tool_call.function.name, "content": json.dumps({"success": False, "error": str(e)}) }) # Step 5: 将所有技能执行结果，连同原始消息，再次发送给LLM进行总结 final_response = await self.llm_client.chat.completions.create( model="gpt-4-turbo", messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_query}, {"role": "assistant", "content": response.choices[0].message.content}, *[ {"role": "tool", "tool_call_id": tc.id, "name": tc.function.name, "content": r["content"]} for tc, r in zip(tool_calls, results) ] ] ) return final_response.choices[0].message.content def _build_tools_description(self) -> str: """生成人类可读的工具描述，用于System Prompt""" descriptions = [] for skill_name, skill_cls in self.registry._skills.items(): desc = f"- `{skill_name}`: {skill_cls.description}" descriptions.append(desc) return "\n".join(descriptions) def _build_tools_schema(self) -> list: """生成OpenAPI格式的tools schema，用于LLM的Function Calling""" schemas = [] for skill_name, skill_cls in self.registry._skills.items(): schema = { "type": "function", "function": { "name": skill_name, "description": skill_cls.description, "parameters": { "type": "object", "properties": {}, "required": [] } } } # 动态填充properties，基于Pydantic模型的schema input_schema = skill_cls.input_model.model_json_schema() schema["function"]["parameters"] = input_schema schemas.append(schema) return schemas # 使用示例 async def main(): # 初始化依赖 http_client = httpx.AsyncClient() registry = SkillRegistry() registry.register(FlightSearchSkill(http_client=http_client)) llm_client = AsyncOpenAI(api_key="your-api-key") agent = SimpleAgent(registry=registry, llm_client=llm_client) # 运行Agent result = await agent.run("帮我查一下明天从北京飞上海的航班，越便宜越好") print(result) if __name__ == "__main__": asyncio.run(main())

这个SimpleAgent虽然只有100多行，但它完整展现了pydantic-ai-skills的集成精髓。最关键的环节是_build_tools_schema()方法——它利用Pydantic v2的model_json_schema()方法，自动将你的FlightSearchInput模型转换成OpenAI Function Calling所需的JSON Schema。这意味着，你修改了FlightSearchInput里的一个字段，LLM的调用参数就会自动同步更新，完全不需要手动去改一堆JSON模板。这种“模型即Schema”的理念，是Pydantic生态最强大的生产力杠杆。我曾负责一个拥有23个技能的客服Agent，当客户要求给所有技能增加一个"language"参数时，我只用了15分钟，就在所有输入模型里加了一行language: str = Field(default="zh-CN")，然后一键部署，所有LLM调用都自动生效了。

4.3 生产环境部署：Docker化与配置管理

当技能从本地开发走向生产，配置管理和环境隔离就成了头等大事。pydantic-ai-skills本身是纯Python库，没有任何特殊部署要求，但为了最佳实践，我推荐一个经过生产验证的Docker化方案：

# Dockerfile FROM python:3.11-slim # 设置工作目录 WORKDIR /app # 复制依赖文件（先复制requirements.txt，利用Docker layer cache） COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY . . # 创建非root用户（安全最佳实践） RUN addgroup -g 1001 -f app && adduser -S app -u 1001 # 切换到非root用户 USER app # 暴露端口（如果需要提供HTTP API） EXPOSE 8000 # 启动命令 CMD ["uvicorn", "my_project.api:app", "--host", "0.0.0.0:8000", "--port", "8000"]

对应的requirements.txt应该这样写，以确保版本精确可控：

# requirements.txt pydantic>=2.5.0,<3.0.0 httpx>=0.25.0 redis>=4.6.0 tenacity>=8.2.0 openai>=1.12.0 uvicorn>=0.23.0

提示：在生产环境中，绝对不要在代码里硬编码API密钥或数据库密码。你应该使用环境变量，并通过Pydantic的BaseSettings来管理。例如：

from pydantic import BaseSettings class Settings(BaseSettings): OPENAI_API_KEY: str REDIS_URL: str = "redis://localhost:6379/0" AIRLINE_API_BASE_URL: str AIRLINE_API_KEY: str class Config: env_file = ".env" # 从.env文件加载 case_sensitive = False settings = Settings()

然后在Docker启动时，通过-e参数传入：

docker run -d \ -e OPENAI_API_KEY=sk-xxx \ -e REDIS_URL=redis://prod-redis:6379/0 \ -e AIRLINE_API_KEY=airline-key-xxx \ --name my-agent \ my-agent-image

这种方式，既保证了密钥的安全，又让不同环境（dev/staging/prod）可以使用同一份镜像，只需改变环境变量即可。

5. 常见问题与排查技巧实录