PydanticAI 是一個(gè)強(qiáng)大的 Python 框架，旨在簡(jiǎn)化使用生成式 AI 的生產(chǎn)級(jí)應(yīng)用程序的開(kāi)發(fā)。它由廣泛使用的數(shù)據(jù)驗(yàn)證庫(kù) Pydantic 背后的同一團(tuán)隊(duì)構(gòu)建，旨在將 FastAPI 的創(chuàng)新和人體工程學(xué)設(shè)計(jì)帶入人工智能應(yīng)用程序開(kāi)發(fā)領(lǐng)域。 PydanticAI 專(zhuān)注于類(lèi)型安全、模塊化以及與其他 Python 工具的無(wú)縫集成。

核心概念

PydanticAI 圍繞幾個(gè)關(guān)鍵概念：

代理商

代理是與大型語(yǔ)言模型 (LLM) 交互的主要接口。代理充當(dāng)各種組件的容器，包括：

• 系統(tǒng)提示：LLM說(shuō)明，定義為靜態(tài)字符串或動(dòng)態(tài)函數(shù)。
• 函數(shù)工具：LLM 可以調(diào)用以獲取其他信息或執(zhí)行操作的函數(shù)。
• 結(jié)構(gòu)化結(jié)果類(lèi)型：LLM 在運(yùn)行結(jié)束時(shí)必須返回的數(shù)據(jù)類(lèi)型。
• 依賴類(lèi)型：系統(tǒng)提示函數(shù)、工具和結(jié)果驗(yàn)證器可能使用的數(shù)據(jù)或服務(wù)。
• LLM 模型：代理將使用的 LLM，可以在代理創(chuàng)建時(shí)或運(yùn)行時(shí)設(shè)置。

代理專(zhuān)為可重用性而設(shè)計(jì)，通常實(shí)例化一次并在整個(gè)應(yīng)用程序中重用。

系統(tǒng)提示

系統(tǒng)提示是開(kāi)發(fā)者向LLM提供的說(shuō)明。他們可以是：

• 靜態(tài)系統(tǒng)提示：在創(chuàng)建代理時(shí)定義，使用 Agent 構(gòu)造函數(shù)的 system_prompt 參數(shù)。
• 動(dòng)態(tài)系統(tǒng)提示：由@agent.system_prompt修飾的函數(shù)定義。它們可以通過(guò) RunContext 對(duì)象訪問(wèn)運(yùn)行時(shí)信息，例如依賴項(xiàng)。

單個(gè)代理可以使用靜態(tài)和動(dòng)態(tài)系統(tǒng)提示，這些提示按照運(yùn)行時(shí)定義的順序附加。

from pydantic_ai import Agent, RunContext
from datetime import date

agent = Agent(
    'openai:gpt-4o',
    deps_type=str,
    system_prompt="Use the customer's name while replying to them.",
)

@agent.system_prompt
def add_the_users_name(ctx: RunContext[str]) -> str:
    return f"The user's name is {ctx.deps}."

@agent.system_prompt
def add_the_date() -> str:
    return f'The date is {date.today()}.'

result = agent.run_sync('What is the date?', deps='Frank')
print(result.data)
#> Hello Frank, the date today is 2032-01-02.

功能工具

功能工具使法學(xué)碩士能夠訪問(wèn)外部信息或執(zhí)行系統(tǒng)提示本身不可用的操作。工具可以通過(guò)多種方式注冊(cè)：

• @agent.tool 裝飾器：適用于需要通過(guò) RunContext 訪問(wèn)代理上下文的工具。
• @agent.tool_plain 裝飾器：適用于不需要訪問(wèn)代理上下文的工具。
Agent 構(gòu)造函數(shù)中的
• tools 關(guān)鍵字參數(shù)：可以采用 Tool 類(lèi)的普通函數(shù)或?qū)嵗?，從而更好地控制工具定義。

from pydantic_ai import Agent, RunContext
from datetime import date

agent = Agent(
    'openai:gpt-4o',
    deps_type=str,
    system_prompt="Use the customer's name while replying to them.",
)

@agent.system_prompt
def add_the_users_name(ctx: RunContext[str]) -> str:
    return f"The user's name is {ctx.deps}."

@agent.system_prompt
def add_the_date() -> str:
    return f'The date is {date.today()}.'

result = agent.run_sync('What is the date?', deps='Frank')
print(result.data)
#> Hello Frank, the date today is 2032-01-02.

工具參數(shù)是從函數(shù)簽名中提取的，用于構(gòu)建工具的 JSON 模式。函數(shù)的文檔字符串用于生成工具的描述以及模式中的參數(shù)描述。

依賴關(guān)系

依賴項(xiàng)通過(guò)依賴項(xiàng)注入系統(tǒng)向代理的系統(tǒng)提示、工具和結(jié)果驗(yàn)證器提供數(shù)據(jù)和服務(wù)。依賴項(xiàng)是通過(guò) RunContext 對(duì)象訪問(wèn)的。它們可以是任何 Python 類(lèi)型，但數(shù)據(jù)類(lèi)是管理多個(gè)依賴項(xiàng)的便捷方法。

import random
from pydantic_ai import Agent, RunContext

agent = Agent(
    'gemini-1.5-flash',
    deps_type=str,
    system_prompt=(
        "You're a dice game, you should roll the die and see if the number "
        "you get back matches the user's guess. If so, tell them they're a winner. "
        "Use the player's name in the response."
    ),
)

@agent.tool_plain
def roll_die() -> str:
    """Roll a six-sided die and return the result."""
    return str(random.randint(1, 6))

@agent.tool
def get_player_name(ctx: RunContext[str]) -> str:
    """Get the player's name."""
    return ctx.deps

dice_result = agent.run_sync('My guess is 4', deps='Anne')
print(dice_result.data)
#> Congratulations Anne, you guessed correctly! You're a winner!

結(jié)果

結(jié)果是代理運(yùn)行返回的最終值。它們包裝在 RunResult（用于同步和異步運(yùn)行）或 StreamedRunResult（用于流式運(yùn)行）中，提供對(duì)使用數(shù)據(jù)和消息歷史記錄的訪問(wèn)。結(jié)果可以是純文本或結(jié)構(gòu)化數(shù)據(jù)，并使用 Pydantic 進(jìn)行驗(yàn)證。

from dataclasses import dataclass
import httpx
from pydantic_ai import Agent, RunContext

@dataclass
class MyDeps:
    api_key: str
    http_client: httpx.AsyncClient

agent = Agent(
    'openai:gpt-4o',
    deps_type=MyDeps,
)

@agent.system_prompt
async def get_system_prompt(ctx: RunContext[MyDeps]) -> str:
    response = await ctx.deps.http_client.get(
        'https://example.com',
        headers={'Authorization': f'Bearer {ctx.deps.api_key}'},
    )
    response.raise_for_status()
    return f'Prompt: {response.text}'

async def main():
    async with httpx.AsyncClient() as client:
        deps = MyDeps('foobar', client)
        result = await agent.run('Tell me a joke.', deps=deps)
        print(result.data)
        #> Did you hear about the toothpaste scandal? They called it Colgate.

通過(guò) @agent.result_validator 裝飾器添加的結(jié)果驗(yàn)證器提供了一種添加進(jìn)一步驗(yàn)證邏輯的方法，特別是當(dāng)驗(yàn)證需要 IO 并且是異步時(shí)。

主要特點(diǎn)

PydanticAI 擁有多項(xiàng)關(guān)鍵功能，使其成為人工智能應(yīng)用程序開(kāi)發(fā)的絕佳選擇：

• 模型不可知論：PydanticAI 支持各種 LLM，包括 OpenAI、Anthropic、Gemini、Ollama、Groq 和 Mistral。它還提供了一個(gè)簡(jiǎn)單的接口來(lái)實(shí)現(xiàn)對(duì)其他模型的支持。
• 類(lèi)型安全：旨在與 mypy 和 Pyright 等靜態(tài)類(lèi)型檢查器無(wú)縫協(xié)作。它允許對(duì)依賴項(xiàng)和結(jié)果類(lèi)型進(jìn)行類(lèi)型檢查。
• 以 Python 為中心的設(shè)計(jì)：利用熟悉的 Python 控制流和代理組合來(lái)構(gòu)建 AI 項(xiàng)目，從而輕松應(yīng)用標(biāo)準(zhǔn) Python 實(shí)踐。
• 結(jié)構(gòu)化響應(yīng)：使用 Pydantic 驗(yàn)證和構(gòu)建模型輸出，確保一致的響應(yīng)。
• 依賴注入系統(tǒng)：提供依賴注入系統(tǒng)，為代理的組件提供數(shù)據(jù)和服務(wù)，增強(qiáng)可測(cè)試性和迭代開(kāi)發(fā)。
• 流式響應(yīng)：支持流式 LLM 輸出并立即驗(yàn)證，從而獲得快速、準(zhǔn)確的結(jié)果。

與代理商合作

運(yùn)行代理

代理可以通過(guò)多種方式運(yùn)行：

• run_sync()：用于同步執(zhí)行。
• run()：用于異步執(zhí)行。
• run_stream()：用于流式響應(yīng)。

from pydantic_ai import Agent, RunContext
from datetime import date

agent = Agent(
    'openai:gpt-4o',
    deps_type=str,
    system_prompt="Use the customer's name while replying to them.",
)

@agent.system_prompt
def add_the_users_name(ctx: RunContext[str]) -> str:
    return f"The user's name is {ctx.deps}."

@agent.system_prompt
def add_the_date() -> str:
    return f'The date is {date.today()}.'

result = agent.run_sync('What is the date?', deps='Frank')
print(result.data)
#> Hello Frank, the date today is 2032-01-02.

對(duì)話

代理運(yùn)行可能代表整個(gè)對(duì)話，但對(duì)話也可以由多次運(yùn)行組成，特別是在維護(hù)交互之間的狀態(tài)時(shí)。您可以使用 message_history 參數(shù)傳遞先前運(yùn)行的消息以繼續(xù)對(duì)話。

import random
from pydantic_ai import Agent, RunContext

agent = Agent(
    'gemini-1.5-flash',
    deps_type=str,
    system_prompt=(
        "You're a dice game, you should roll the die and see if the number "
        "you get back matches the user's guess. If so, tell them they're a winner. "
        "Use the player's name in the response."
    ),
)

@agent.tool_plain
def roll_die() -> str:
    """Roll a six-sided die and return the result."""
    return str(random.randint(1, 6))

@agent.tool
def get_player_name(ctx: RunContext[str]) -> str:
    """Get the player's name."""
    return ctx.deps

dice_result = agent.run_sync('My guess is 4', deps='Anne')
print(dice_result.data)
#> Congratulations Anne, you guessed correctly! You're a winner!

使用限制

PydanticAI 提供了一個(gè) settings.UsageLimits 結(jié)構(gòu)來(lái)限制令牌和請(qǐng)求的數(shù)量。您可以通過(guò) use_limits 參數(shù)將這些設(shè)置應(yīng)用到運(yùn)行函數(shù)。

from dataclasses import dataclass
import httpx
from pydantic_ai import Agent, RunContext

@dataclass
class MyDeps:
    api_key: str
    http_client: httpx.AsyncClient

agent = Agent(
    'openai:gpt-4o',
    deps_type=MyDeps,
)

@agent.system_prompt
async def get_system_prompt(ctx: RunContext[MyDeps]) -> str:
    response = await ctx.deps.http_client.get(
        'https://example.com',
        headers={'Authorization': f'Bearer {ctx.deps.api_key}'},
    )
    response.raise_for_status()
    return f'Prompt: {response.text}'

async def main():
    async with httpx.AsyncClient() as client:
        deps = MyDeps('foobar', client)
        result = await agent.run('Tell me a joke.', deps=deps)
        print(result.data)
        #> Did you hear about the toothpaste scandal? They called it Colgate.

模型設(shè)置

settings.ModelSettings 結(jié)構(gòu)允許您通過(guò)溫度、max_tokens 和超時(shí)等參數(shù)微調(diào)模型行為。您可以通過(guò)運(yùn)行函數(shù)中的 model_settings 參數(shù)應(yīng)用這些。

from pydantic import BaseModel
from pydantic_ai import Agent

class CityLocation(BaseModel):
    city: str
    country: str

agent = Agent('gemini-1.5-flash', result_type=CityLocation)
result = agent.run_sync('Where were the olympics held in 2012?')
print(result.data)
#> city='London' country='United Kingdom'

功能工具詳解

工具注冊(cè)

可以使用@agent.tool裝飾器（對(duì)于需要上下文的工具）、@agent.tool_plain裝飾器（對(duì)于沒(méi)有上下文的工具）或通過(guò)Agent構(gòu)造函數(shù)中的tools參數(shù)來(lái)注冊(cè)工具。

from pydantic_ai import Agent

agent = Agent('openai:gpt-4o')

# Synchronous run
result_sync = agent.run_sync('What is the capital of Italy?')
print(result_sync.data)
#> Rome

# Asynchronous run
async def main():
    result = await agent.run('What is the capital of France?')
    print(result.data)
    #> Paris

    async with agent.run_stream('What is the capital of the UK?') as response:
        print(await response.get_data())
        #> London

工具架構(gòu)

參數(shù)描述從文檔字符串中提取并添加到工具的 JSON 架構(gòu)中。如果工具具有可以表示為 JSON 模式中的對(duì)象的單個(gè)參數(shù)，則該模式將簡(jiǎn)化為該對(duì)象。

from pydantic_ai import Agent

agent = Agent('openai:gpt-4o', system_prompt='Be a helpful assistant.')
result1 = agent.run_sync('Tell me a joke.')
print(result1.data)
#> Did you hear about the toothpaste scandal? They called it Colgate.

result2 = agent.run_sync('Explain?', message_history=result1.new_messages())
print(result2.data)
#> This is an excellent joke invent by Samuel Colvin, it needs no explanation.

動(dòng)態(tài)工具

可以使用準(zhǔn)備函數(shù)來(lái)自定義工具，該函數(shù)在每個(gè)步驟中調(diào)用以修改工具定義或從該步驟中省略工具。

from pydantic_ai import Agent
from pydantic_ai.settings import UsageLimits
from pydantic_ai.exceptions import UsageLimitExceeded

agent = Agent('claude-3-5-sonnet-latest')
try:
    result_sync = agent.run_sync(
        'What is the capital of Italy? Answer with a paragraph.',
        usage_limits=UsageLimits(response_tokens_limit=10),
    )
except UsageLimitExceeded as e:
    print(e)
    #> Exceeded the response_tokens_limit of 10 (response_tokens=32)

消息和聊天記錄

訪問(wèn)消息

代理運(yùn)行期間交換的消息可以通過(guò) RunResult 和 StreamedRunResult 對(duì)象上的 all_messages() 和 new_messages() 方法訪問(wèn)。

from pydantic_ai import Agent

agent = Agent('openai:gpt-4o')
result_sync = agent.run_sync(
    'What is the capital of Italy?',
    model_settings={'temperature': 0.0},
)
print(result_sync.data)
#> Rome

消息重用

消息可以傳遞到 message_history 參數(shù)以在多個(gè)代理運(yùn)行之間繼續(xù)對(duì)話。當(dāng)message_history設(shè)置且不為空時(shí)，不會(huì)生成新的系統(tǒng)提示。

消息格式

消息格式與模型無(wú)關(guān)，允許消息在不同的代理中使用，或者與使用不同模型的同一代理一起使用。

調(diào)試與監(jiān)控

派丹提克原火

PydanticAI 與 Pydantic Logfire 集成，這是一個(gè)可觀察平臺(tái)，可讓您監(jiān)控和調(diào)試整個(gè)應(yīng)用程序。 Logfire 可用于：

• 實(shí)時(shí)調(diào)試：實(shí)時(shí)查看應(yīng)用程序中發(fā)生的情況。
• 監(jiān)控應(yīng)用程序性能：使用 SQL 查詢和儀表板。

要將 PydanticAI 與 Logfire 一起使用，請(qǐng)使用 logfire 可選組進(jìn)行安裝：pip install 'pydantic-ai[logfire]'。然后，您需要配置 Logfire 項(xiàng)目并驗(yàn)證您的環(huán)境。

安裝和設(shè)置

安裝

PydanticAI 可以使用 pip 安裝：

from pydantic_ai import Agent, RunContext
from datetime import date

agent = Agent(
    'openai:gpt-4o',
    deps_type=str,
    system_prompt="Use the customer's name while replying to them.",
)

@agent.system_prompt
def add_the_users_name(ctx: RunContext[str]) -> str:
    return f"The user's name is {ctx.deps}."

@agent.system_prompt
def add_the_date() -> str:
    return f'The date is {date.today()}.'

result = agent.run_sync('What is the date?', deps='Frank')
print(result.data)
#> Hello Frank, the date today is 2032-01-02.

還可以使用精簡(jiǎn)安裝來(lái)使用特定型號(hào)，例如：

import random
from pydantic_ai import Agent, RunContext

agent = Agent(
    'gemini-1.5-flash',
    deps_type=str,
    system_prompt=(
        "You're a dice game, you should roll the die and see if the number "
        "you get back matches the user's guess. If so, tell them they're a winner. "
        "Use the player's name in the response."
    ),
)

@agent.tool_plain
def roll_die() -> str:
    """Roll a six-sided die and return the result."""
    return str(random.randint(1, 6))

@agent.tool
def get_player_name(ctx: RunContext[str]) -> str:
    """Get the player's name."""
    return ctx.deps

dice_result = agent.run_sync('My guess is 4', deps='Anne')
print(dice_result.data)
#> Congratulations Anne, you guessed correctly! You're a winner!

日志火集成

要將 PydanticAI 與 Logfire 一起使用，請(qǐng)使用 logfire 可選組安裝它：

from dataclasses import dataclass
import httpx
from pydantic_ai import Agent, RunContext

@dataclass
class MyDeps:
    api_key: str
    http_client: httpx.AsyncClient

agent = Agent(
    'openai:gpt-4o',
    deps_type=MyDeps,
)

@agent.system_prompt
async def get_system_prompt(ctx: RunContext[MyDeps]) -> str:
    response = await ctx.deps.http_client.get(
        'https://example.com',
        headers={'Authorization': f'Bearer {ctx.deps.api_key}'},
    )
    response.raise_for_status()
    return f'Prompt: {response.text}'

async def main():
    async with httpx.AsyncClient() as client:
        deps = MyDeps('foobar', client)
        result = await agent.run('Tell me a joke.', deps=deps)
        print(result.data)
        #> Did you hear about the toothpaste scandal? They called it Colgate.

示例

示例可作為單獨(dú)的包提供：

from pydantic import BaseModel
from pydantic_ai import Agent

class CityLocation(BaseModel):
    city: str
    country: str

agent = Agent('gemini-1.5-flash', result_type=CityLocation)
result = agent.run_sync('Where were the olympics held in 2012?')
print(result.data)
#> city='London' country='United Kingdom'

測(cè)試與評(píng)估

單元測(cè)試

單元測(cè)試驗(yàn)證您的應(yīng)用程序代碼的行為是否符合預(yù)期。對(duì)于 PydanticAI，請(qǐng)遵循以下策略：

• 使用 pytest 作為您的測(cè)試工具。
• 使用 TestModel 或 FunctionModel 代替您的實(shí)際模型。
• 使用 Agent.override 替換應(yīng)用程序邏輯中的模型。
• 全局設(shè)置 ALLOW_MODEL_REQUESTS=False 以防止意外調(diào)用非測(cè)試模型。

from pydantic_ai import Agent

agent = Agent('openai:gpt-4o')

# Synchronous run
result_sync = agent.run_sync('What is the capital of Italy?')
print(result_sync.data)
#> Rome

# Asynchronous run
async def main():
    result = await agent.run('What is the capital of France?')
    print(result.data)
    #> Paris

    async with agent.run_stream('What is the capital of the UK?') as response:
        print(await response.get_data())
        #> London

埃瓦爾斯

評(píng)估用于衡量 LLM 的表現(xiàn)，更像是基準(zhǔn)測(cè)試而不是單元測(cè)試。評(píng)估的重點(diǎn)是衡量法學(xué)碩士在特定申請(qǐng)中的表現(xiàn)。這可以通過(guò)端到端測(cè)試、綜合獨(dú)立測(cè)試、使用 LLM 評(píng)估 LLM 或通過(guò)測(cè)量生產(chǎn)中的代理性能來(lái)完成。

示例用例

PydanticAI 可用于多種用例：

• 輪盤(pán)賭輪：使用具有整數(shù)依賴項(xiàng)和布爾結(jié)果的代理模擬輪盤(pán)賭輪。
• 聊天應(yīng)用程序：創(chuàng)建多次運(yùn)行的聊天應(yīng)用程序，使用 message_history 傳遞以前的消息。
• 銀行支持代理：使用工具、依賴項(xiàng)注入和結(jié)構(gòu)化響應(yīng)為銀行構(gòu)建支持代理。
• 天氣預(yù)報(bào)：創(chuàng)建一個(gè)應(yīng)用程序，使用函數(shù)工具和依賴項(xiàng)根據(jù)位置和日期返回天氣預(yù)報(bào)。
• SQL 生成：根據(jù)用戶提示生成 SQL 查詢，并使用結(jié)果驗(yàn)證器進(jìn)行驗(yàn)證。

結(jié)論

PydanticAI 提供了一個(gè)強(qiáng)大而靈活的框架，用于開(kāi)發(fā)人工智能應(yīng)用程序，重點(diǎn)強(qiáng)調(diào)類(lèi)型安全性和模塊化。使用 Pydantic 進(jìn)行數(shù)據(jù)驗(yàn)證和結(jié)構(gòu)化，再加上其依賴注入系統(tǒng)，使其成為構(gòu)建可靠且可維護(hù)的人工智能應(yīng)用程序的理想工具。憑借其廣泛的 LLM 支持以及與 Pydantic Logfire 等工具的無(wú)縫集成，PydanticAI 使開(kāi)發(fā)人員能夠高效構(gòu)建強(qiáng)大的、可用于生產(chǎn)的 AI 驅(qū)動(dòng)項(xiàng)目。

以上是PydanticAI：構(gòu)建生產(chǎn)就緒型人工智能應(yīng)用程序的綜合指南的詳細(xì)內(nèi)容。更多信息請(qǐng)關(guān)注PHP中文網(wǎng)其他相關(guān)文章！