Zenn
🏌️

Pydantic AIでLLMアプリケーションを監視する

に公開
observability
LLM
LLMOps
AIエージェント
PydanticAI
tech

PydanticAIでAgentを作成し、そのAgentが動いているLLMアプリケーションを監視したいと思います。今回の記事ではLLMの単なるWrapperをAgentとして定義し、Pydantic Logfireでやってみます。

本文中のコード: https://github.com/hosimesi/code-for-techblogs/tree/main/pydanticai_observability

PydanitcAIとは

PydanticAIは、生成AIのアプリケーションを作るためのPythonフレームワークです。最近ではAIエージェントが注目されていますが、それらを構築するためのツールを提供します。元々PydanticはFastAPIをはじめとするPythonの静的型付けを提供するツールであり、PydanticAIでも同様に機能します。
PydanticAIには以下のような機能が提供されています。

  • Dependency
    • システムプロンプトやツールに対してDI（Dependency Injection）が可能になります。また、あるAgentを別のエージェントが呼び出すこともできます。
  • Tool
    • 現在はベクトル検索はできませんが、追加する議論があるので要チェックです。
  • Agent
    • LLMを呼び出すための共通のインターフェース。LLMサービスが提供しているストリーミングや構造化出力にも基本的に対応しています。
  • Observability
    • Pydantic Logfireと連携し、LLMアプリケーションの監視も可能になっています。

また、そのほかにもPytestなどのテストツールと統合していたり、MCPクライアントや、ワークフロー形式のAgentを順番に実行できるノードの集合であるグラフ機能なども提供しています。この辺りはまた別記事で紹介したいと思います。
まだメジャーバージョンはリリースされていませんが、2025年の夏を目処にV1がリリースされる予定です。

Pydantic Logfireとは

Pydantic Logfireは生成AIアプリケーション用の可観測性プラットフォームであり、パフォーマンス、リクエスト、使用状況などを監視します。もちろん、生成AIアプリケーションだけでなく、通常のアプリケーションでも使用可能です。つまり、Pydanticを使ったFastAPIアプリケーションのバリデーション結果だけを集めることも可能です。また、Pydantic LogfireはOpenTelemetryの上に構築されているため、素のOpenTelemetry Python SDKを使ってメトリクスを送信することも可能です。
集めたデータに対してクエリを書くことができたり、いくつかのダッシュボードも用意されています。コード上では手動でトレーシングを書くこともでき、少しのコードの追加で簡単にシステム監視が可能になります。
https://pydantic.dev/logfire
※ https://pydantic.dev/logfire

2025年4月末現在では、Pydantic Logfireを使うためにはアカウント登録が必要となっており、フリープランとプロプランがあります。そして、使用するためにプロジェクトを作成しておきます。
プロジェクト作成画面

環境構築

uvを使用して環境を構築します。まずは、必要になるpydantic-aiと監視のためのlogfireをインストールします。

$ uv init

$ uv add pydantic-ai

$ uv add "pydantic-ai[logfire]"

また、環境変数読み込みのためのdotenvもインストールします。

$ uv add python-dotenv

また、LLMとしてGeminiを使用するので.envにGEMINI_API_KEYを設定します。

$ cp .env.example .env

$ vim .env
.env
GEMINI_API_KEY=xxxxx

こちらで準備は完了です。

検証

Agentの作成

今回はトレーシングをメインで検証するため、Agent自体は最小限の機能で作成します。単純にチャット履歴を持ったLLMアプリケーションで、ユーザーが停止するまでCLI上でチャットができるツールになっています。

main.py
from dotenv import load_dotenv
from pydantic_ai import Agent
import logging

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

load_dotenv(verbose=True)


def main():
    agent = Agent(
        "google-gla:gemini-2.0-flash-lite",
        system_prompt="You are a helpful assistant that can search the web for information.",
    )
    history = []

    try:
        while True:
            query = input("Enter a query: ")
            if history:
                result = agent.run_sync(query, message_history=history[-1])
            else:
                result = agent.run_sync(query)
            logger.info(result.new_messages())
            history.append(result.new_messages())
            print(result.output)
    except KeyboardInterrupt:
        print("\nChat has been stopped by user.")


if __name__ == "__main__":
    main()

実行してみます。

$ uv run src/main.py

Enter a query: 世界で一番高い山は？
INFO:httpx:HTTP Request: POST
世界で一番高い山は、エベレスト山です。
Enter a query: 二番目は？
INFO:httpx:HTTP Request: POST
世界で二番目に高い山は、K2（チョゴリ）です。

Pydantic Logfireへのメトリクスの送信

先ほどのアプリケーションに、Logfireのコードを数行ほど足すだけでメトリクスをとることができます。

main.py
import logging

+ import logfire
from dotenv import load_dotenv
from pydantic_ai import Agent

+ logfire.configure()

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

load_dotenv(verbose=True)


def main():
    agent = Agent(
        "google-gla:gemini-2.0-flash-lite",
        system_prompt="You are a helpful assistant that can search the web for information.",
+       instrument=True, // instrumentを取得するようにします
    )
                :
                :

複数のAgentがある場合、一つ一つ設定するのではなく、Agent.instrument_all()を追加するだけで全てのAgentのメトリクスを取得できます。そして、データを送信するためにLogfireの設定を行います。CLI上で以下のコマンドを実行することで、ブラウザ経由でトークンを払い出すことができます。

$ uv run logfire auth

もしくは、コンソールからトークンを取得し、コード上で設定することもできます。

main.py
- logfire.configure()
+ logfire.configure(token=os.getenv("LOGFIRE_TOKEN"))

そして実行していくつかチャットをしてみます。

$ uv run src/main.py

Pydantic Logfireでのメトリクスの確認

ブラウザからLogfireにログインし、メトリクスを送信したプロジェクトを選択します。Logfireには以下のタブが用意されており、確認したいメトリクスに合わせて使い分けることができます。

  • Live
    • 時系列に沿ってリアルタイムにアプリケーションの監視ができます。
  • Dashboards
    • 自身で書いたクエリを実行した結果のビューを組み合わせてダッシュボードを作成できます。
  • Alerts
    • 取得したメトリクスに閾値を設定し、リアルタイムにアラートを上げることができます。
  • Explore
    • クエリを自由に実行して取得することができます。

Liveタブを見てみると、先ほど行ったチャットの結果が一覧で確認できます。一般的なログ監視ツールと同じような使い方ができるのが便利でした。
Liveタブ
また、クエリを書くことができ、メトリクスにタグがいくつも付いているので、それでフィルターをかけることができそうです。例えば、「レスポンスの文字列に〜が含まれているもの」や「〇〇のモデルを使っているもの」などに絞ることができます。
Liveクエリ
Dashboardsタブではゼロから作成したり、公式が元々提供しているダッシュボードも使うことができます。提供ダッシュボードでは一般的なアプリケーションのHTTPステータスを監視できるので、FastAPIにそのまま使うことができそうです。
Dashboardタブ
Alertsタブでは取得したメトリクスの状態の閾値を指定し、アラートを投げることができます。現在は、Webhookが登録できるのでSlackへの通知なども可能で、業務でも使える機能になっています。
Alertsタブ
Exploreタブでは取得したメトリクスに対して自由にクエリを投げられるエディタが用意されています。ここで取得した結果は表形式だけでなく、簡単なグラフもすぐ描画でき、最近のDWHなどとUIの観点ではあまり差分がないくらいの機能は整っています。
Exploreタブ

まとめ

PydanticAIを使ってLLMアプリケーションを監視してみました。LLMアプリケーションは非決定論的なシステムなので監視は必要ですし、今後AI Agentが普及するとよりシステムの複雑性は増すため、非常に便利なツールだと感じました。Pydantic Logfireを用いることで、これらのアプリケーションにおけるトレーシングやメトリクスの監視が容易になり、システムの安定性とパフォーマンスの向上が期待できます。今後、通常のFastAPIなどのアプリケーションへの組み込みやマルチエージェントの監視も試してみようと思います。

参考

https://ai.pydantic.dev/
https://www.ai-shift.co.jp/techblog/5267

Discussion