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

核心概念

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

代理商

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

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

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

系統(tǒng)提示

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

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

單個代理可以使用靜態(tài)和動態(tài)系統(tǒ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.

功能工具

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

• @agent.tool 裝飾器：適用于需要通過 RunContext 訪問代理上下文的工具。
• @agent.tool_plain 裝飾器：適用于不需要訪問代理上下文的工具。
Agent 構(gòu)造函數(shù)中的
• tools 關(guān)鍵字參數(shù)：可以采用 Tool 類的普通函數(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)系

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

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

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

主要特點

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

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

與代理商合作

運行代理

代理可以通過多種方式運行：

• 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.

對話

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

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 提供了一個 settings.UsageLimits 結(jié)構(gòu)來限制令牌和請求的數(shù)量。您可以通過 use_limits 參數(shù)將這些設(shè)置應(yīng)用到運行函數(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)允許您通過溫度、max_tokens 和超時等參數(shù)微調(diào)模型行為。您可以通過運行函數(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'

功能工具詳解

工具注冊

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

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 模式中的對象的單個參數(shù)，則該模式將簡化為該對象。

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.

動態(tài)工具

可以使用準(zhǔn)備函數(shù)來自定義工具，該函數(shù)在每個步驟中調(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)

消息和聊天記錄

訪問消息

代理運行期間交換的消息可以通過 RunResult 和 StreamedRunResult 對象上的 all_messages() 和 new_messages() 方法訪問。

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ù)以在多個代理運行之間繼續(xù)對話。當(dāng)message_history設(shè)置且不為空時，不會生成新的系統(tǒng)提示。

消息格式

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

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

派丹提克原火

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

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

要將 PydanticAI 與 Logfire 一起使用，請使用 logfire 可選組進(jìn)行安裝：pip install 'pydantic-ai[logfire]'。然后，您需要配置 Logfire 項目并驗證您的環(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.

還可以使用精簡安裝來使用特定型號，例如：

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 一起使用，請使用 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.

示例

示例可作為單獨的包提供：

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'

測試與評估

單元測試

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

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

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

埃瓦爾斯

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

示例用例

PydanticAI 可用于多種用例：

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

結(jié)論

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

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