syakesaba

Network/Infrastructure/Data

概要

PydanticAIは大規模言語モデル(以後LLM)を使ったフレームワークツール。今回もGoogle Gemini APIの無料枠を使用する。この記事ではLangfuse.comの無料枠を使用することにより、無料でPydanticAIをトレースできる事を示す。Google Gemini APIではgemini-2.0-flashを使用する。今回もGoogle Colaboratoryを使用する。

手順

大まかな手順はこのページに記載の通り。

Open In Colab

Langfuse.comへの登録とAPIキーの準備

  1. Langfuseにアクセスし、適当なアカウントで登録。Free Planなのでクレカ情報などは要らない。
  2. defaultという名前でOrganizationを作成。メンバーは自分だけ。
  3. default-projectという名前でProjectを作成。
  4. その後の画面で出てくる、Secret Key, Public Key, Hostを全てメモする。

Google Colaboratoryの準備

  1. Colaboratoryを起動し、以下通りSecretを入力する。
    • GeminiのAPIキーをGOOGLE_API_KEYとして入力
    • LangfuseのPublic KeyをLANGFUSE_PUBLIC_KEYとして入力
    • LangfuseのSecret KeyをLANGFUSE_SECRET_KEYとして入力
    • LangfuseのHostをLANGFUSE_HOSTとして入力
  2. その後、以下のようにしてSecretを読み込み、環境変数に登録する。
!pip install -qq pydantic_ai[logfire] nest_asyncio
import os
import base64
import nest_asyncio
nest_asyncio.apply() # Google Colab自体がasyncio配下で動いているのでネストさせる。
import logfire
from google.colab import userdata
GOOGLE_API_KEY = userdata.get("GOOGLE_API_KEY") # Secrets（🔑）からGOOGLE_API_KEYをNotebook access可能にする
LANGFUSE_PUBLIC_KEY = userdata.get("LANGFUSE_PUBLIC_KEY") # Secrets（🔑）からLANGFUSE_PUBLIC_KEYをNotebook access可能にする
LANGFUSE_SECRET_KEY = userdata.get("LANGFUSE_SECRET_KEY") # Secrets（🔑）からLANGFUSE_SECRET_KEYをNotebook access可能にする
LANGFUSE_HOST = userdata.get("LANGFUSE_HOST") # Secrets（🔑）からLANGFUSE_HOSTをNotebook access可能にする

# GeminiModelに関する環境変数をGoogle Colabに作成
os.environ["GEMINI_API_KEY"] = GOOGLE_API_KEY

# OTLP_EXPORTERに関する環境変数をGoogle Colabに作成
# https://langfuse.com/docs/integrations/pydantic-ai
LANGFUSE_AUTH = base64.b64encode(f"{LANGFUSE_PUBLIC_KEY}:{LANGFUSE_SECRET_KEY}".encode()).decode()
os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = f"{LANGFUSE_HOST}/api/public/otel"
os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = f"Authorization=Basic {LANGFUSE_AUTH}"

# logfire cloudへの余計な送信を無効化
import logfire
_ = logfire.configure(
    service_name='MyGoogleColab',
    # Sending to Logfire is on by default regardless of the OTEL env vars.
    send_to_logfire=False,
)
  • PydanticAIもlogfireもデフォルトがasyncによる非同期関数だらけなので、nest_asyncio.apply()しないとうまいことJupyterNotebookで動かない。

PydanticAIの利用

import random
from typing import Annotated, Optional
from pydantic_ai import Agent, RunContext
from pydantic_ai.models.gemini import GeminiModel
from pydantic import BaseModel, StringConstraints
from dataclasses import dataclass

model = GeminiModel("gemini-2.0-flash")

@dataclass
class JankenResult:
    result: Annotated[str, StringConstraints(pattern="^(グー|チョキ|パー)$")] | None

@dataclass
class JankenResults:
    mario: JankenResult
    luigi: JankenResult

mario = Agent(
    model=model, name="Mario",
    system_prompt="あなたはジャンケンをするAIエージェントです。"
                  "'get_janken' toolを使用し、ジャンケンで「グー」、「チョキ」、「パー」のどれかをランダムに出します。",
    deps_type=None, result_type=JankenResult, instrument=True
)
luigi = Agent(
    model=model, name="Luigi",
    system_prompt="あなたはジャンケンをするAIエージェントです。"
                  "'get_janken' toolを使用し、ジャンケンで「グー」、「チョキ」、「パー」のどれかをランダムに出します。",
    deps_type=None, result_type=JankenResult, instrument=True
)
peach = Agent(
    model=model, name="Peach",
    system_prompt="あなたはジャンケン勝負の結果を出力するAIエージェントです。"
                  "'prompt_janken' toolを使用し、ジャンケン勝負の結果と勝敗を出力してください。",
    deps_type=JankenResults, result_type=str, instrument=True
)

yoshi = Agent(
    model=model, name="Yoshi",
    system_prompt="あなたは親切なAIエージェントです。"
                  "get_janken_results' toolを使用し結果を教えてください。",
    deps_type=None, result_type=str, instrument=True
)

@mario.tool
def get_janken(ctx: RunContext[None]) -> str:
    return random.choice(["グー", "チョキ", "パー"])

@luigi.tool
def get_janken(ctx: RunContext[None]) -> str:
    return random.choice(["グー", "チョキ", "パー"])

@peach.tool
async def prompt_janken(ctx: RunContext[None]) -> str:
    mario_response = await mario.run(
        f'ジャンケン、',
        usage=ctx.usage,
    )
    luigi_response = await luigi.run(
        f'ジャンケン、',
        usage=ctx.usage,
    )
    jankenResults = JankenResults(mario=mario_response.data, luigi=luigi_response.data)
    print(jankenResults)
    if jankenResults.mario.result == "グー" and jankenResults.luigi.result == "グー":
        return "双方グーで引き分けです。"
    elif jankenResults.mario.result == "グー" and jankenResults.luigi.result == "チョキ":
        return "Marioがグーで勝ちです。"
    elif jankenResults.mario.result == "グー" and jankenResults.luigi.result == "パー":
        return "Luigiがパーで勝ちです。"
    elif jankenResults.mario.result == "チョキ" and jankenResults.luigi.result == "グー":
        return "Luigiがグーで勝ちです。"
    elif jankenResults.mario.result == "チョキ" and jankenResults.luigi.result == "チョキ":
        return "双方チョキで引き分けです。"
    elif jankenResults.mario.result == "チョキ" and jankenResults.luigi.result == "パー":
        return "Marioがチョキで勝ちです。"
    elif jankenResults.mario.result == "パー" and jankenResults.luigi.result == "グー":
        return "Marioがパーで勝ちです。"
    elif jankenResults.mario.result == "パー" and jankenResults.luigi.result == "チョキ":
        return "Luigiがチョキで勝ちです。"
    elif jankenResults.mario.result == "パー" and jankenResults.luigi.result == "パー":
        return "双方パーで引き分けです。"
    else:
        return "不明な結果です。"

@yoshi.tool
async def get_janken_results(ctx: RunContext[None]) -> str:
    response = await peach.run("ジャンケン勝負の結果と勝敗を出力してください", usage=ctx.usage)
    return response.data

result = await yoshi.run("どちらがどの手を出したかと、勝敗を教えて下さい。", deps=None, result_type=str)
print(result.data) # マリオがチョキを出して勝ちました。
  • Agentを作成する際にinstrument=Trueを引数に与えるとトレースが送信される。

トレースの参照

https://cloud.langfuse.com/を開き、default Organizationからdefault-project Projectを開いて、Tracesを開く。

langfuse_cloud_trace.png

ここまでトレースできると楽しいですね。

関連記事