用 Pydantic AI 生成带客户上下文的模拟客服结果

一个客服代理需要同时处理三类信息：应用已经确认的客户身份、数据库可查询的业务数据，以及模型组织出的回答。Pydantic AI 的官方 Bank Support 示例用一个内存 SQLite 数据库，将这三部分放进很短的程序：动态指令补充客户姓名，工具提供余额，Pydantic 模型约束最终输出的字段。

本文译编自 Pydantic AI 官方 Bank Support 示例，维护与发布方为 Pydantic，原页未单列可确认的个人作者。2026 年 10 月 5 日读取的页面使用 instructions、output_type 和 result.output。它是教学用的模拟客服，没有停卡接口，也没有客户认证与账户授权；本文未运行程序或调用付费模型。

三个接口各负责什么

SupportDependencies 保存 customer_id 与数据库包装对象。每次调用 run_sync 时，应用把它作为 deps 传进去。动态指令和工具函数都通过 RunContext[SupportDependencies] 取得同一份依赖。

@support_agent.instructions 注册的函数会查询姓名，并返回一段附加指令，使模型知道如何称呼客户。@support_agent.tool 注册的余额工具则查询该客户余额，格式化为两位小数的美元字符串。模型最终交回的不是任意 Python 对象，而是 SupportOutput，其中包括文字建议 support_advice、布尔值 block_card 和整数 risk。

这个区分很重要：block_card=True 是一个结构化结果字段，不是调用银行系统后的执行回执。Pydantic 的类型验证可以约束结果形状，但不会自动证明回答真实、业务动作完成或风险判断正确。

运行入口与版本条件

官方 示例安装说明 要求安装可选的 examples 依赖组，例如 pip install "pydantic-ai[examples]" 或 uv add "pydantic-ai[examples]"；克隆仓库时使用 uv sync --extra examples。完成所选模型提供商的认证配置后，原文使用以下入口：

python -m pydantic_ai_examples.bank_support
# 或：
uv run -m pydantic_ai_examples.bank_support

原文代码指定 openai:gpt-5.2，页面还给出 PYDANTIC_AI_MODEL=gemini-3-flash-preview ... 的运行提示。但展示的单文件代码没有显式读取这个环境变量；直接复制代码时，应自行核对模型选择如何生效，不能假定任意入口都支持覆盖。模型名称、可用性与 SDK 接口会变，应锁定所用库版本并查阅对应文档。

配置认证会使调用具备访问模型服务的能力，也可能产生费用。本文没有写入或读取真实密钥。只用虚构客户数据练习，因为姓名、余额与对话会进入模型请求；内存数据库并不意味着这些内容始终留在本机。

完整示例：从虚构客户到结构化结果

下面保留原页的程序结构与行为，翻译说明性注释并把示例输出移到后文解释。特意保留数据库包装类中对全局 cur 的引用，便于与原文核对；其可复用性问题在下一节给出修改办法。

import sqlite3
from dataclasses import dataclass

from pydantic import BaseModel
from pydantic_ai import Agent, RunContext


@dataclass
class DatabaseConn:
    """SQLite 连接的包装类。"""

    sqlite_conn: sqlite3.Connection

    async def customer_name(self, *, id: int) -> str | None:
        res = cur.execute('SELECT name FROM customers WHERE id=?', (id,))
        row = res.fetchone()
        if row:
            return row[0]
        return None

    async def customer_balance(self, *, id: int) -> float:
        res = cur.execute('SELECT balance FROM customers WHERE id=?', (id,))
        row = res.fetchone()
        if row:
            return row[0]
        else:
            raise ValueError('Customer not found')


@dataclass
class SupportDependencies:
    customer_id: int
    db: DatabaseConn


class SupportOutput(BaseModel):
    support_advice: str
    """给客户的客服建议。"""
    block_card: bool
    """是否建议停卡；字段本身不会执行停卡。"""
    risk: int
    """查询的风险级别。"""


support_agent = Agent(
    'openai:gpt-5.2',
    deps_type=SupportDependencies,
    output_type=SupportOutput,
    instructions=(
        'You are a support agent in our bank, give the '
        'customer support and judge the risk level of their query. '
        "Reply using the customer's name."
    ),
)


@support_agent.instructions
async def add_customer_name(ctx: RunContext[SupportDependencies]) -> str:
    customer_name = await ctx.deps.db.customer_name(id=ctx.deps.customer_id)
    return f"The customer's name is {customer_name!r}"


@support_agent.tool
async def customer_balance(ctx: RunContext[SupportDependencies]) -> str:
    """返回客户当前账户余额。"""
    balance = await ctx.deps.db.customer_balance(
        id=ctx.deps.customer_id,
    )
    return f'${balance:.2f}'


if __name__ == '__main__':
    with sqlite3.connect(':memory:') as con:
        cur = con.cursor()
        cur.execute('CREATE TABLE customers(id, name, balance)')
        cur.execute("""
            INSERT INTO customers VALUES
                (123, 'John', 123.45)
        """)
        con.commit()

        deps = SupportDependencies(
            customer_id=123,
            db=DatabaseConn(sqlite_conn=con),
        )
        result = support_agent.run_sync('What is my balance?', deps=deps)
        print(result.output)

        result = support_agent.run_sync('I just lost my card!', deps=deps)
        print(result.output)

执行入口使用 sqlite3.connect(':memory:') 建立临时数据库，创建 customers 表，再插入编号为 123、姓名为 John、余额为 123.45 的虚构客户。示例在模块级主入口内创建 cur，因此按这个入口运行时，前面的异步方法能够找到该全局游标；这不代表数据库类已经正确封装了连接。

两个查询都使用 SQL 参数占位符 ?，将客户 ID 作为参数传递；这比把 ID 拼接到 SQL 字符串中更可靠。余额工具没有让模型选择另一个账户 ID，而是从应用依赖中读取。不过，应用本身仍必须保证依赖中的客户 ID 来自已验证的登录主体，不能直接相信请求参数。

函数写成 async def，也不意味着 SQLite 操作会自动变为非阻塞 I/O。这里调用的是同步 sqlite3 接口；对教学数据量问题有限，扩展到并发服务时需要另行设计连接管理与执行方式。

怎样解读原文展示的两个结果

原页在余额请求后展示了一个包含 block_card=False、risk=1 的结果，回复称 John 的余额为 123.45 美元，并加上“包括待处理交易”的表述。随后在“我刚丢了银行卡”的请求后，示例结果给出 block_card=True、risk=8，并声称正在临时停卡。

这两处都必须按代码能力来解读：数据库只有 balance 字段，没有区分可用、账面或待处理交易余额，所以“包括待处理交易”没有得到这个示例的数据支持；程序也没有任何停卡工具、接口调用或持久化更新，因此“正在停卡”不是已完成的实际动作。这些是原文展示的模型输出，不是本次执行结果，也不是每次调用都会得到的固定答案。

把全局游标改为实例连接

原类声明了 sqlite_conn 字段，但两个查询实际上都使用模块全局的 cur。把类移到其他模块、创建多个连接或并发使用时，这会造成未初始化引用或串用连接。下面是编辑补充的局部修正版：通过实例保存的连接执行查询，并在姓名不存在时明确失败，避免把 None 当作客户姓名写入指令。

@dataclass
class DatabaseConn:
    sqlite_conn: sqlite3.Connection

    async def customer_name(self, *, id: int) -> str:
        row = self.sqlite_conn.execute(
            'SELECT name FROM customers WHERE id=?', (id,)
        ).fetchone()
        if row is None:
            raise ValueError('Customer not found')
        return row[0]

    async def customer_balance(self, *, id: int) -> float:
        row = self.sqlite_conn.execute(
            'SELECT balance FROM customers WHERE id=?', (id,)
        ).fetchone()
        if row is None:
            raise ValueError('Customer not found')
        return row[0]

这项修改解决的是连接归属和缺失姓名的处理方式，仍然没有补齐认证、事务、并发管理或资金精度。原文用 float 表示金额，显示成两位小数不能让二进制浮点变成精确货币运算；真实业务应按其币种与会计规则选用整数最小货币单位或十进制定点方案。

风险字段与模型指令都需要业务边界

原文的 risk: int 没有限定范围，也没有定义风险量表。若教学应用明确采用 0 到 10 的整数等级，可以作如下编辑扩展示例；这不是原文已经具备的约束：

from pydantic import BaseModel, Field

class SupportOutput(BaseModel):
    support_advice: str
    block_card: bool
    risk: int = Field(ge=0, le=10)

范围验证只拦截越界数字，不保证模型的风险评估可靠。实际停卡应由独立业务流程完成身份与权限检查、确认操作条件、调用明确接口，并根据真实回执生成客户话术。这个示例应使用“建议申请停卡”一类表达，避免在没有动作回执时声称成功。

还有一条与 SQL 注入不同的边界：姓名被格式化后直接加入模型的动态指令。!r 只生成 Python 字符串表示，并不能阻止模型把恶意姓名中的文字理解为指令。客户消息、数据库文本和工具结果都应当被视为业务数据，不能因为进入了指令字符串就变成可信规则。这里没有针对提示注入的实际防御验证。

本示例展示了动态指令、工具和结构化输出如何配合，适合用虚构数据理解框架。它没有提供银行级身份验证、授权、金额处理、操作审计或错误恢复。静态审查没有看到硬编码 API 密钥，但这不等于代码无漏洞；本文也没有安装依赖、测试局部修正版或向模型发送请求。

The MIT License (MIT)

Copyright (c) Pydantic Services Inc. 2024 to present

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.