Zenn
Finatext Tech BlogPublicationへの投稿
💭

Pydantic AI でエージェントを作ってみる

Ryotaro
2024/12/10に公開
Pydantic
agent
生成 AI
記事投稿コンテスト「今年の最も大きなチャレンジ」
pydanticai
tech

alt text

この記事は、Finatextグループ Advent Calendar 2024の10日目の記事です。

はじめに

こんにちは、ナウキャストでLLMエンジニアをしているRyotaroです。
毎週生成 AI に関する情報をまとめていますが、今回はアドベントカレンダー期間ということで久々にテーマを一つに絞って書きたいと思います！

そのテーマというのが、「Pydantic AI」です！
最近 bolt とか replit とか AI エージェントを駆使したサービスがめちゃくちゃ流行っていますよね。そのエージェントを作るためのライブラリの一つとして Pydantic AI が最近出たみたいなので紹介したいと思います。

Pydantic AI とは

Pydantic AI は、Python を用いて大規模言語モデル（LLM）を活用するためのエージェントフレームワークです。このフレームワークは、Pydantic を基盤にしており、型安全性や構造化レスポンス、エージェントやツール呼び出し、依存性注入などの機能を提供します。

Pydantic AI の特徴は以下の通りです。

  • 型安全な構造化レスポンス: LLMからの生のテキスト出力をPydanticモデルで定義することができ、これにより構造化データとして受け取ることが可能です。バリデーションに失敗した場合には再試行が促されます。
  • モデル非依存アーキテクチャ: OpenAIやGoogleのGeminiなど、複数のLLMモデルをプラグイン的に利用できる設計になっています。（ただし現状は OpenAI と Gemini、Groq のみで、Claude は今後対応するそうです）
  • ツール呼び出し: LLMに対して関数一覧を渡すことで、必要な情報を取得するための関数呼び出しを簡潔に実装できます。これにより、高度な情報取得（RAG）やエージェント的な挙動が可能になります。
  • 依存性注入: システムプロンプトやツール関数は外部リソースから受け取ることができ、テスト性や拡張性が向上します。
  • ストリーミング対応: LLMの結果をリアルタイムで処理することもサポートされています。

サンプルコードはこんな感じです。

from pydantic_ai import Agent

def roulette_wheel(ctx: RunContext[int], square: int) -> str:
    """check if the square is a winner"""
    return 'winner' if square == ctx.deps else 'loser'

agent = Agent(
    'openai:gpt-4o',
    deps_type=int,
    result_type=str,
    tools=[roulette_wheel],
    system_prompt='You are a roulette wheel. You will be given a number and you will need to check if it is a winner.',
)

result = agent.run_sync("I want to bet on the number 1", deps=1)
print(result.data)
#> 'loser'

後ほど詳しく記述していますが、他のライブラリに比べて入出力の型を明確に指定できるのが Pydantic AI の良い点です。

Pydantic AI の準備

さて、本題の Pydantic AI ですが、ここからは詳しくまずはインストールとセットアップから行きます。

Installation

install は簡単で、pip や poetry で pydantic-ai をインストールするだけです。

poetry add pydantic-ai

また、パッケージごとにもインストールすることができます。

# OpenAI のみインストール
poetry add pydantic-ai-slim[openai]

# Gemini のみインストール
# Gemini だけは `pydantic-ai-slim[gemini]` ではなく `pydantic-ai-slim` としてください
poetry add pydantic-ai-slim

# Groq のみインストール
poetry add pydantic-ai-slim[groq]

# Logfire のみインストール
poetry add pydantic-ai-slim[logfire]

Setup

Pydantic AI では現在 OpenAI、Gemini、Groq のモデルが使える状態で、Anthropic のモデルも今後使えるようになる予定です。

OpenAI と Gemini、Groq はそれぞれ API key を設定するだけで使えるようになります。なぜか gemini だけはモデル名だけで指定するようです。

from pydantic_ai import Agent

agent_openai = Agent('openai:gpt-4o')
agent_gemini = Agent('gemini-1.5-flash')
agent_groq = Agent('groq:llama-3.1-70b-versatile')

ちなみに、API を直接設定したい場合はそれぞれのモデルに対応したクライアントを作成してあげる必要があります。

モデルごとのコード

OpenAI の場合

from openai import AsyncOpenAI

from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIModel

client = AsyncOpenAI(api_key='your-api-key')
model = OpenAIModel('gpt-4o', openai_client=client)
agent = Agent(model)

Gemini の場合

from pydantic_ai import Agent
from pydantic_ai.models.gemini import GeminiModel

model = GeminiModel('gemini-1.5-flash', api_key='your-api-key')
agent = Agent(model)

Groq の場合

from pydantic_ai import Agent
from pydantic_ai.models.groq import GroqModel

model = GroqModel('llama-3.1-70b-versatile', api_key='your-api-key')
agent = Agent(model)

Azure OpenAI の場合はもう少し工夫が必要で、OpenAI の AsyncOpenAI を使ってクライアントを作成してあげる必要があります。

from openai import AsyncAzureOpenAI

from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIModel

client = AsyncAzureOpenAI(
    azure_endpoint='...',
    api_version='2024-07-01-preview',
    api_key='your-api-key',
)

model = OpenAIModel('deployment-name', openai_client=client)
agent = Agent(model)

また、Vertex AI の場合は service account の json ファイルを指定してあげる必要があります。

from pydantic_ai import Agent
from pydantic_ai.models.vertexai import VertexAIModel

model = VertexAIModel(
    'gemini-1.5-flash',
    service_account_file='path/to/service-account.json',
    region='asia-east1',
)
agent = Agent(model)

Pydantic AI の基本的な使い方

Pydantic AI の主なコンポーネントは以下の通りです。

  • Agents
    • Pydantic AI の主要インターフェースで、モデルを指定してエージェントを作成します。
    • モデル以外も、後述の Dependency や Tool 、Result Type、System Prompt を指定してエージェントを作成します。
  • Dependencies
    • エージェントに依存するモデルを指定します。
    • これによりエージェントが受け取るメッセージ以外の情報の型を指定することができます。
  • Result Type
    • エージェントの結果の型を指定します。
    • これによりエージェントの結果の型を指定することができます。
  • System Prompt
    • エージェントのシステムプロンプトを指定します。
    • エージェント作成時でも、デコレーターでも指定できます。
  • Tools
    • エージェントに依存するツールを指定します。
    • エージェント作成時でも、デコレーターでも指定できます。
    • これによりエージェントが実行できるツールを指定することができます。

Agents

Agent は Pydantic AI の主要インターフェースで、モデルを指定してエージェントを作成します。

from pydantic_ai import Agent

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

Agent のコンストラクタは以下の通りです。

class Agent(
    model: KnownModelName | Model | None = None,
    *,
    result_type: type[str] = str,
    system_prompt: str | Sequence[str] = (),
    deps_type: type[User] = NoneType,
    name: str | None = None,
    retries: int = 1,
    result_tool_name: str = 'final_result',
    result_tool_description: str | None = None,
    result_retries: int | None = None,
    tools: Sequence[Tool[User] | ToolFuncContext[User, ...] | ToolFuncPlain[...]] = (),
    defer_model_check: bool = False
)

基本は model・deps_type・result_type・system_prompt を指定してエージェントを作成して、tools をあとから追加するという流れかなと思います。

name はエージェントの名前でログを出すときに使われます。何も指定しない場合は初回の実行時に自動的に生成されるそうです。

name: str | None
"""The name of the agent, used for logging.
If `None`, we try to infer the agent name from the call frame when the agent is first run.
"""

実行する時は run か run_sync か run_async を使います。（run_sync は内部的には asyncio.run(self.run()) を呼び出すだけです）
基本的には run_sync か run_async を使うことになると思います。

result = agent.run_sync('Hello, world!') # 同期実行
result = agent.run_async('Hello, world!') # 非同期実行

Dependencies

Dependencies はエージェント作成時に型を指定することで、エージェントに依存するデータを指定することができます。

from pydantic_ai import Agent

agent = Agent('openai:gpt-4o', deps_type=int)
result = agent.run_sync('Hello, world!', deps=1)
print(result.data)
#> True

また、その依存性注入した値は system prompt や tools の中で ctx.deps とアクセスできます。
例えば以下のように、tools の中で依存性注入した値を使って処理を行うことができます。（後述しますが、system_prompt や tools 関数の引数の初めの型は RunContext というクラスにする必要があります。）

from pydantic_ai import Agent

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

@agent.tool
async def add_one(ctx: RunContext[int]) -> int:
    return ctx.deps + 1 # 依存性注入した値（deps=1）を使って処理を行う

result = agent.run_sync('Hello, world!', deps=1)
print(result.data)
#> 2

面白いのが、依存関係の中に別のエージェントを指定することもできます。
次の例では、joke_agent が factory_agent に依存していて、Tell me a joke というと、joke_factory というツールを使って内部で factory_agent を実行し複数のジョークを生成。その上で、joke_agent はその中から最も良いジョークを選んで返しています。

from dataclasses import dataclass
from pydantic_ai import Agent, RunContext

@dataclass
class MyDeps:
    factory_agent: Agent[None, list[str]]

joke_agent = Agent(
    'openai:gpt-4o',
    deps_type=MyDeps,
    system_prompt=(
        'Use the "joke_factory" to generate some jokes, then choose the best. '
        'You must return just a single joke.'
    ),
)

factory_agent = Agent('gemini-1.5-pro', result_type=list[str])

@joke_agent.tool
async def joke_factory(ctx: RunContext[MyDeps], count: int) -> str:
    r = await ctx.deps.factory_agent.run(f'Please generate {count} jokes.')
    return '\n'.join(r.data)

result = joke_agent.run_sync('Tell me a joke.', deps=MyDeps(factory_agent))
print(result.data)
#> Did you hear about the toothpaste scandal? They called it Colgate.

Result Type

Result Type はエージェントの結果の型を指定します。（あとから指定できないので、Agent 作成時に指定する必要があります。）また Agent の実際の返り値は RunResult というクラスで、data に result_type で指定した型の結果が入っています。

from pydantic_ai import Agent

agent = Agent('openai:gpt-4o')
result = agent.run_sync('Hello, world!')
print(result.data)
#> Hello! How can I assist you today?

それ以外にも cost や messages などの情報が入っています。

from pydantic_ai import Agent

agent = Agent('openai:gpt-4o')
result = agent.run_sync('Hello, world!')
print(result.cost())
#> cost = Cost(
#>     request_tokens=11,
#>     response_tokens=9,
#>     total_tokens=20,
#>     details={
#>         'accepted_prediction_tokens': 0,
#>         'audio_tokens': 0,
#>         'reasoning_tokens': 0,
#>         'rejected_prediction_tokens': 0,
#>         'cached_tokens': 0
#>     }
#> )
print(result.all_messages())
#> messages = [
#>     UserPrompt(
#>         content='Hello, world!',
#>         timestamp=datetime.datetime(2024, 12, 9, 5, 54, 34, 937902, tzinfo=datetime.timezone.utc),
#>         role='user'
#>     ),
#>     ModelTextResponse(
#>         content='Hello! How can I assist you today?',
#>         timestamp=datetime.datetime(2024, 12, 9, 5, 54, 35, tzinfo=datetime.timezone.utc),
#>         role='model-text-response'
#>     )
#> ]

さらに、 I/O の検証を設定することも簡単にできます。
次の例では、Response として Success か InvalidRequest のいずれかを返すようにしています。
そしてデコレーターで validate_result という関数を定義して、InvalidRequest の場合はその結果を、Success の場合は実際に sql を実行し、その結果 QueryError が発生した場合は ModelRetry を発生させ、成功したらその結果を返すようにしています。

from typing import Union

from pydantic import BaseModel
from pydantic_ai import Agent, RunContext, ModelRetry

from fake_database import DatabaseConn, QueryError

class Success(BaseModel):
    sql_query: str

class InvalidRequest(BaseModel):
    error_message: str

Response = Union[Success, InvalidRequest]
agent: Agent[DatabaseConn, Response] = Agent(
    'gemini-1.5-flash',
    result_type=Response,  # type: ignore
    deps_type=DatabaseConn,
    system_prompt='ユーザー入力に基づいてPostgreSQL風のSQLクエリを生成してください。',
)

@agent.result_validator
async def validate_result(ctx: RunContext[DatabaseConn], result: Response) -> Response:
    if isinstance(result, InvalidRequest):
        return result
    try:
        await ctx.deps.execute(f'EXPLAIN {result.sql_query}')
    except QueryError as e:
        raise ModelRetry(f'Invalid query: {e}') from e
    else:
        return result

result = agent.run_sync(
    '昨日アクティブだったユーザーを取得してください。', deps=DatabaseConn()
)
print(result.data)
#> sql_query='SELECT * FROM users WHERE last_active::date = today() - interval 1 day'

System Prompt

system prompt はエージェント作成時に指定することもできますし、デコレーターで指定することもできます。

from datetime import date

from pydantic_ai import Agent, RunContext

agent = Agent(
    'openai:gpt-4o',
    deps_type=str,
    system_prompt="顧客の名前を使って返信してください。",
)

@agent.system_prompt
def add_the_users_name(ctx: RunContext[str]) -> str:
    return f"ユーザーの名前は {ctx.deps} です。"

@agent.system_prompt
def add_the_date() -> str:
    return f'今日の日付は {date.today()} です。'

result = agent.run_sync('今日の日付は何ですか？', deps='Frank')
print(result.data)
#> 今日の日付は 2024-12-09 です。

前述したように、デコレーター system_prompt では引数を指定するときに初めは RunContext を引数に取る必要があります。そうすることで、依存性注入したデータをを system prompt に使うことができます。

Tools

tools はエージェントに依存するツールを指定します。ツールは非同期関数として定義し、RunContext を引数に取り、deps に依存性注入したデータを使って処理を行います。エージェントにツールを登録する方法はいくつかあります。

  • デコレータ経由 @agent.tool - エージェントコンテキストへのアクセスが必要なツール用
  • デコレータ経由 @agent.tool_plain - エージェントコンテキストへのアクセスを必要としないツール用
  • tools キーワード引数を介して Agent、単純な関数または Tool を渡す

@agent.tool ほとんどの場合、ツールはエージェント コンテキストにアクセスする必要があるため、デフォルトになっています。

import random

from pydantic_ai import Agent, RunContext

agent = Agent(
    'gemini-1.5-flash',
    deps_type=str,
    system_prompt=(
        "あなたはサイコロゲームです。サイコロを振って出た数字がユーザーの予想と一致するか確認してください。"
        "一致した場合は、ユーザーに勝者であることを伝えてください。"
        "返答にはプレイヤーの名前を使用してください。"
    ),
)

@agent.tool_plain
def roll_die() -> str:
    """サイコロを振って出た数字を返してください。"""
    return str(random.randint(1, 6))

@agent.tool
def get_player_name(ctx: RunContext[str]) -> str:
    """ユーザーの名前を返してください。"""
    return ctx.deps

dice_result = agent.run_sync('私の予想は4です', deps='Anne')
print(dice_result.data)
#> サイコロを振って出た数字は 4 です。
#> あなたは勝者です！

さらに、この Tools の優れているところとして関数から docstring を抽出して、ツールの description を自動で連携してくれるところです。


from pydantic_ai import Agent
from pydantic_ai.messages import Message, ModelAnyResponse, ModelTextResponse
from pydantic_ai.models.function import AgentInfo, FunctionModel

agent = Agent()

@agent.tool_plain
def foobar(a: int, b: str, c: dict[str, list[float]]) -> str:
    """Get me foobar.

    Args:
        a: apple pie
        b: banana cake
        c: carrot smoothie
    """
    return f'{a} {b} {c}'

def print_schema(messages: list[Message], info: AgentInfo) -> ModelAnyResponse:
    tool = info.function_tools[0]
    print(tool.description)
    #> Get me foobar.
    print(tool.parameters_json_schema)
    """
    {
        'description': 'Get me foobar.',
        'properties': {
            'a': {'description': 'apple pie', 'title': 'A', 'type': 'integer'},
            'b': {'description': 'banana cake', 'title': 'B', 'type': 'string'},
            'c': {
                'additionalProperties': {'items': {'type': 'number'}, 'type': 'array'},
                'description': 'carrot smoothie',
                'title': 'C',
                'type': 'object',
            },
        },
        'required': ['a', 'b', 'c'],
        'type': 'object',
        'additionalProperties': False,
    }
    """
    return ModelTextResponse(content='foobar')

agent.run_sync('hello', model=FunctionModel(print_schema))

Pydantic AI でエージェントを作る

ではせっかくなので DB を操作するエージェントを作ってみます。

まずは DB のテーブルを作成します。

-- 顧客情報テーブル: customers
-- 顧客ID, 名前, 連絡先
CREATE TABLE customers (
    customer_id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    contact TEXT,
);

-- サンプルデータの挿入
-- customers テーブルのサンプルデータ
INSERT INTO customers (customer_id, name, contact) VALUES
(1, '山田太郎', 'taro@example.com'),
(2, '佐藤花子', 'hanako@example.com'),
(3, '田中一郎', 'ichiro@example.com');

sqlite3 でテーブルを作成します。

sqlite3 test.db < test.sql

次に、Agent の Dependencies 用に DBConnection を定義します。

import sqlite3
from typing import Any, List, Tuple

class DBConnection:

    def __init__(self, db_path: str):
        self.db_path = db_path
        self.conn = None
        self.cursor = None

    def connect(self):
        """データベースへ接続し、カーソルオブジェクトを取得します。"""
        if self.conn is None:
            self.conn = sqlite3.connect(self.db_path)
            self.cursor = self.conn.cursor()

    def execute_query(self, query: str, params: Tuple[Any, ...] = ()) -> List[Tuple]:
        """
        SELECT系クエリを実行して結果を返します。
        paramsはプレースホルダ付きクエリに渡すパラメータタプルです。

        例:
        results = db.execute_query("SELECT * FROM customers WHERE budget < ?", (150000,))
        """
        if self.conn is None:
            self.connect()
        self.cursor.execute(query, params)
        result = self.cursor.fetchall()
        self.close()
        return result

    def execute_non_query(self, query: str, params: Tuple[Any, ...] = ()) -> int:
        """
        INSERT, UPDATE, DELETEなど、戻り値が不要なクエリを実行します。
        実行後にcommitを行い、変更された行数を返します。

        例:
        affected_rows = db.execute_non_query("UPDATE properties SET price=? WHERE property_id=?", (90000, 1))
        """
        if self.conn is None:
            self.connect()
        self.cursor.execute(query, params)
        self.conn.commit()
        result = self.cursor.rowcount
        self.close()
        return result

    def close(self):
        """データベースコネクションを閉じます。"""
        if self.conn is not None:
            self.conn.close()
            self.conn = None
            self.cursor = None

一応テストします。

db = DBConnection('test.db')
db.connect()
results = db.execute_query('SELECT * FROM customers')
print(results)
#> [(1, '山田太郎', 'taro@example.com'), (2, '佐藤花子', 'hanako@example.com'), (3, '田中一郎', 'ichiro@example.com')]
db.close()

これで DB に接続できるようになりました。

そしたらいよいよ Agent を作成します。

from typing import Union

from openai import AsyncAzureOpenAI
from pydantic import BaseModel
from pydantic_ai import Agent, RunContext, ModelRetry
from pydantic_ai.models.openai import OpenAIModel

client = AsyncAzureOpenAI(
    azure_endpoint='...',
    api_version='2024-07-01-preview',
    api_key='your-api-key',
)
model = OpenAIModel('gpt-4o-2024-08-06', openai_client=client)

db_agent = Agent(
    model,
    deps_type=DBConnection, # 先ほど作成した DBConnection を依存性注入する
    result_type=str,
    system_prompt=
        'ユーザー入力に基づいてPostgreSQL風のSQLクエリを生成してください'
        'もしユーザーからの情報が不足していた場合は、その情報をユーザーに追加で聞いてください'
)

@db_agent.tool
async def fetch_all_customers(ctx: RunContext[DBConnection]) -> str:
    """顧客情報を全て取得"""
    result = ctx.deps.execute_query('SELECT * FROM customers')
    return result

@db_agent.tool
async def add_customer(ctx: RunContext[DBConnection], name: str, contact: str) -> str:
    """顧客情報を追加
    Args:
        name: 顧客名
        contact: 顧客連絡先
    """
    result = ctx.deps.execute_non_query(f'INSERT INTO customers (name, contact) VALUES ("{name}", "{contact}")')
    return result

では、まずは顧客情報を全て取得してみます。

result = db_agent.run_sync('顧客情報を全て取得してください', deps=DBConnection('test.db'))
print(result.data)
#> [(1, '山田太郎', 'taro@example.com'), (2, '佐藤花子', 'hanako@example.com'), (3, '田中一郎', 'ichiro@example.com')]

これで顧客情報を全て取得できました。

次に、顧客情報を追加してみます。

result = db_agent.run_sync('顧客情報を追加してください', deps=DBConnection('test.db'))
print(result.data)
#>以下の顧客情報を教えてください：
#>
#>1. 顧客名
#>2. 顧客連絡先

すると、顧客情報を追加するための情報が足りないという回答が返ってきます。
なので、あらためて顧客情報を含めて再度実行します。

result = db_agent.run_sync(
    '顧客情報を追加してください。顧客名は"test"、連絡先は"test@example.com"としてください。',
    deps=DBConnection('test.db'),
)
print(result.data)
#>顧客情報が正常に追加されました。顧客名は"test"、連絡先は"test@example.com"です。

これで顧客情報を追加できました。
再度顧客情報を全て取得してみます。

result = db_agent.run_sync('顧客情報を全て取得してください', deps=DBConnection('test.db'))
print(result.data)
#> [(1, '山田太郎', 'taro@example.com'), (2, '佐藤花子', 'hanako@example.com'), (3, '田中一郎', 'ichiro@example.com'), (4, 'test', 'test@example.com')]

これで顧客情報を全て取得できました。

Pydantic AI まとめ

Pydantic AI を使ってみて、めちゃくちゃ簡単にエージェントを作ることができました。Pydantic 由来なので公式でも記載の通り FastAPI の router を作るときと同じように Agent を定義して作ることができるのは嬉しいですね。
型チェックも効くので、エラーが出るときはそのエラーを解消することでエージェントの挙動を改善することができます。
今回は DB を操作するエージェントを作ってみましたが、外部 API Connection を定義すれば、その API を操作するエージェントも作ることができます。汎用性は結構高いですね。

結論としては、ライトにエージェントを作りたいという方には Pydantic AI が結構良いんじゃないかなと思いました。まだ version が 0.0.12（2024-12-10）なので今後の update にも期待していきたいと思います！

他の AI Agent ライブラリ

ちなみに他にも Agent を構築できるライブラリとして LangChain Agents や LangGraph や Praison AIなどがあります。

LangChain Agents

LangChain v0.0.1 の機能で、AgentExecutor というもので Agent を構築します。ただ、LangChain コミュニティでは段々と LangGraph に移行しているようです。

実際にはこんな感じで書きます。

from langchain import hub
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_community.tools.tavily_search import TavilySearchResults

llm = ChatOpenAI(model="gpt-4o", temperature=0)
search = TavilySearchResults()
prompt = hub.pull("hwchase17/openai-functions-agent")

agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

agent_executor.invoke({"input": "hi!"})

LangGraph

LangChain の Agent から、グラフの概念を取り入れて Agent を構築するように拡張して作られたライブラリです。

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, StateGraph, START
from langgraph.prebuilt import tools_condition

builder = StateGraph(State)

builder.add_node("assistant", Assistant(part_1_assistant_runnable))
builder.add_node("tools", create_tool_node_with_fallback(part_1_tools))

builder.add_edge(START, "assistant")
builder.add_conditional_edges(
    "assistant",
    tools_condition,
)
builder.add_edge("tools", "assistant")

part_1_graph = builder.compile()

そして以下のコードで構築したグラフを描画できます。

from IPython.display import Image, display

display(Image(part_1_graph.get_graph().draw_mermaid_png()))

alt text

グラフを可視化できるのは LangGraph の特徴ですね。

Praison AI

Praison AI は Agent の役割を yaml 形式で定義して、それを取り込んで Agent を構築するようにしています。

まずはtool となる関数を作成します。

from duckduckgo_search import DDGS
from praisonai_tools import BaseTool

class InternetSearchTool(BaseTool):
    name: str = "Internet Search Tool"
    description: str = "Search Internet for relevant information based on a query or latest news"

    def _run(self, query: str):
        ddgs = DDGS()
        results = ddgs.text(keywords=query, region='wt-wt', safesearch='moderate', max_results=5)
        return results

次にプレイブックという yaml ファイルを作成して、Agent の役割を定義します。

framework: crewai
topic: research about the causes of lung disease
roles:
  research_analyst:
    backstory: Experienced in analyzing scientific data related to respiratory health.
    goal: Analyze data on lung diseases
    role: Research Analyst
    tasks:
      data_analysis:
        description: Gather and analyze data on the causes and risk factors of lung diseases.
        expected_output: Report detailing key findings on lung disease causes.
    tools:
    - InternetSearchTool

そして以下のコードで Agent を構築します。

praisonai = PraisonAI(agent_yaml=agent_yaml)
result = praisonai.run()

yaml で静的なファイルとして Agent の定義を作成できるのは Praison AI の特徴ですね。

比較してみて

他のライブラリと比べると、Pydantic AI は LangChain Agent と似ているという感じで、もう LangGraph に移行しつつあることを考慮するとサポートされていかない可能性があるので Pydantic AI が良いと思います。LangGraph はグラフを可視化できる点が他と差別化されているポイントですが、構築がやや煩雑なイメージがありますね。

参考資料

https://ai.pydantic.dev/
https://github.com/pydantic/pydantic-ai
https://zenn.dev/kun432/scraps/83e7570920a802
https://zenn.dev/zenkigen_tech/articles/536801e61d0689
https://qiita.com/Tadataka_Takahashi/items/764421590550a7af765e

We are Hiring!

この件についてもっと詳しく知りたい、議論したい、はたまたナウキャストという会社に興味を持ったという方は、カジュアル面談フォームから連絡ください。

https://herp.careers/v1/finatexthd/vZWzSlI_B-qk

Finatext Tech Blog
Finatext Tech BlogPublication

『金融を"サービス"として再発明する』Finatextグループのテックブログです。

Discussion