在人工智能快速发展的今天，智能代理(Agentic AI) 成为了一个热门话题。本文将介绍如何使用 PydanticAI 构建智能代理系统，包括与 MCP(Model Context Protocol) 的集成以及一个实际的智能工单助手应用案例。

PydanticAI 与 MCP 简介

PydanticAI 是一个强大的 Python 库，它允许开发者以类型安全的方式定义和使用 AI 代理。它与 Pydantic v2 完美集成，提供了清晰的数据验证和模式定义能力。而 MCP(Model Context Protocol) 则是一个标准化协议，用于定义 LLM 如何与工具交互。

MCP 的优势

  • 标准化的工具定义方式
  • 可重用性强
  • 支持使用预构建或自定义服务器暴露工具
  • 避免在使用新工具时重复造轮子

实战案例：构建智能工单助手

让我们通过构建一个端到端的智能工单助手系统来深入了解 PydanticAI 的实际应用。

环境准备

首先，我们需要安装必要的依赖：

1
2
3

pip install --upgrade pip
pip install pydantic-ai
pip install nest_asyncio  # 用于处理异步操作

数据模型定义

使用 Pydantic v2 定义清晰的数据模型：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

from openai import AsyncAzureOpenAI

from dataclasses import dataclass
from typing import Literal
from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext
from pydantic_ai.models.openai import OpenAIModel
from pydantic_ai.providers.openai import OpenAIProvider


class CreateTicketOutput(BaseModel):
    ticket_id: str = Field(..., description="唯一工单标识符")
    summary: str = Field(..., description="问题摘要")
    severity: Literal["low", "medium", "high"] = Field(..., 
                                            description="紧急程度")
    department: str = Field(..., description="负责部门")
    status: Literal["open"] = Field("open", description="初始工单状态")

class TicketStatusOutput(BaseModel):
    ticket_id: str = Field(..., description="唯一工单标识符")
    status: Literal["open", "in_progress", "resolved"] = Field(..., 
                                            description="当前工单状态")

数据库设置

使用 SQLite 作为轻量级数据存储：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

import sqlite3
import uuid

conn = sqlite3.connect(":memory:")
conn.execute("""
    CREATE TABLE tickets (
        ticket_id TEXT PRIMARY KEY,
        summary TEXT NOT NULL,
        severity TEXT NOT NULL,
        department TEXT NOT NULL,
        status TEXT NOT NULL
    )
""")
conn.commit()

智能代理实现

创建两个智能代理，分别用于创建工单和查询状态：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

@dataclass
class TicketingDependencies:
    """用于系统提示和工具的数据库连接"""
    db: sqlite3.Connection

# 创建工单代理
client = AsyncAzureOpenAI(
    azure_endpoint='your-azure-endpoint',
    api_key='your-api-key',
    api_version='2025-04-01-preview',
)

model = OpenAIModel(
    'gpt-4o',
    provider=OpenAIProvider(openai_client=client),
)

create_agent = Agent(
    model=model,
    deps_type=TicketingDependencies,
    output_type=CreateTicketOutput,
    system_prompt="您是一个工单助手。使用`create_ticket`工具记录新问题。"
)

@create_agent.tool
async def create_ticket(
    ctx: RunContext[TicketingDependencies],
    summary: str,
    severity: Literal["low", "medium", "high"],
    department: str
) -> CreateTicketOutput:
    """在数据库中记录新工单"""
    tid = str(uuid.uuid4())
    ctx.deps.db.execute(
        "INSERT INTO tickets VALUES (?,?,?,?,?)",
        (tid, summary, severity, department, "open")
    )
    ctx.deps.db.commit()
    return CreateTicketOutput(
        ticket_id=tid,
        summary=summary,
        severity=severity,
        department=department,
        status="open"
    )

使用示例

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# 创建依赖实例
deps = TicketingDependencies(db=conn)

# 创建新工单
create_result = await create_agent.run(
    "三楼打印机显示卡纸错误。",
    deps=deps
)
print("创建的工单 →")
print(create_result.output.model_dump_json(indent=2))

系统优势

  1. 类型安全：通过 Pydantic 模型确保数据验证和类型检查
  2. 模块化设计：清晰的代理职责划分
  3. 可扩展性：易于添加新功能和集成其他服务
  4. 自然语言交互：支持通过自然语言创建和查询工单
  5. 数据一致性：强制执行数据模式和业务规则

未来展望

这个系统还可以进一步扩展：

  1. 集成更多 AI 模型提供商
  2. 添加更复杂的工单处理流程
  3. 实现多代理协作
  4. 增加更多自定义工具
  5. 集成外部API和服务

结论

通过 PydanticAI 和 MCP 的结合，我们可以构建出既强大又可维护的智能代理系统。这种方法不仅提供了类型安全和数据验证，还支持灵活的扩展和集成。随着智能代理技术的不断发展，这种架构模式将变得越来越重要。

无论是构建客服系统、自动化工作流，还是其他智能应用，PydanticAI 都提供了一个可靠的基础。通过类型安全的代理定义和清晰的数据模型，我们可以构建出更可靠、更易维护的 AI 应用。