- BrainTools - https://www.braintools.ru -

Агенты в Pydantic AI от вызова LLM до MCP

Введение

Всем привет, сегодня я расскажу вам о том, как делать можно делать агентов с помощью Pydantic AI.
Pydantic AI – фреймворк от создателей Pydantic – популярной библиотеки для валидации данных в Python с ядром на Rust.

Начнем с простых примеров в виде вызова LLM , а затем постепенно будем усложнять задачу, создавая более сложного агента.

Виртуальное окружение.
В качестве пакетного менеджера в данном проекте используется uv [1], однако вы можете использовать любой другой, удобный вам, просто имейте это ввиду при установке пакетов.

Репозиторий
Весь код примеров ниже доступен на github [2]

Модель
В качестве стартовой LLM я предлагаю использовать Mistral [3], а потом переехать на что то более умное, но тоже бесплатное.
Если у вас нет токена, идем на сайт, проходим регистрацию и получаем токен, делаем Research план, который дает бесплатно много токенов.
Альтернативно: можно попробовать поднять локально что то через Ollama, если ваши ресурсы позволяют, однако дойдя до более сложной имплементации вы столкнетесь с тем, что агент не может решить проблему, поэтому гонять запросы в API проще.

Переменные
Создаем .env файл, в нем пишем, например:

  MISTRAL_API_KEY=ваш_токен

Зависимости на старте

uv add pydantic-ai python-dotenv

Начнем с простого

Дергаем LLM на простой вопрос
Попробуем просто дернуть Mistral API для вызова модельки и ответа на самый простой вопрос.

from pydantic_ai import Agent
from pydantic_ai.models.mistral import MistralModel
from pydantic_ai.providers.mistral import MistralProvider
import os
from dotenv import load_dotenv
load_dotenv()
model = MistralModel(model_name="mistral-large-latest",    
                     provider=MistralProvider(
                        api_key=os.getenv("MISTRAL_API_KEY")),
)
agent = Agent(model=model, retries=5)
result_sync = agent.run_sync("Столица Гондураса?")
print(result_sync.data)
#Столица Гондураса — это город Тегусигальпа.

Как мы видим принцип написания вызова LLM немного отличается от того, что предлагает Langchain, а также значительно проще чем через Open AI SDK или чистый Mistral API.

Стоит отметить что с точки зрения [4] Pydantic AI агент – это просто интерфейс взаимодействия с LLM, что отвечает в рамках данного фреймворка на вопрос о том, что-же такое этот ваш агент.

Ваш агент может состоять просто из вызова модели, можно добавить к нему системный промпт, формат вывода и другие параметры и это будет все тот же агент, просто с доп функционалом.

Агенты в Pydantic AI от вызова LLM до MCP - 1

А вообще в документации [5] прекрасно описано что такое агент и что он делает, поэтому едем дальше.

Logfire (опционально)

Для трассировки жизненного пути агента, можно воспользоваться Pydantic Logfire [6]. Это позволит нам подробнее посмотреть что делает LLM когда вызывает какой-то тул, например. Ну или просто историю сообщений.
Logfire еще не такой продвинутый как тот же LangSmith или Langfuse, однако его просто поставить и с ним просто работать, особенно полезно, чтоб понять нужна ли нам та тула, которую мы только что придумалиWeather Model мы уже видели, а вот Supervisor Model лучше знать че она может получить, поэтому перечислим как опциональные модели от предыдуших действительно, или итак все норм работает.
Кстати там есть дашборды как и у конкурентов, но лично у меня они не показывают абсолютно ничего. А сидеть и самому написывать SQL-запросы как-то лень и не входит в задачу данной статьи.
Поэтому. проходим по ссылке выше, проходим регистрацию, создаем проект и наблюдаем за нашим агентом.

Вернемся к нашему агенту.

uv add logfire

Добавляем токен через конфигурацию logfire и добавляем аргумент instrument=True к агенту.

from pydantic_ai import Agent
from pydantic_ai.models.mistral import MistralModel
from pydantic_ai.providers.mistral import MistralProvider
import logfire
import os
from dotenv import load_dotenv
load_dotenv()
logfire.configure(send_to_logfire=os.getenv("LOGFIRE_TOKEN"))
model = MistralModel(model_name="mistral-large-latest",
                      provider=MistralProvider
                      (api_key=os.getenv("MISTRAL_API_KEY")),
)
agent = Agent(model=model, retries=5, instrument=True)
result_sync = agent.run_sync("Столица Гондураса?")
print(result_sync.data)

Structured Output

Удобная штука, которая заставит модель выдать именно то, что вам надо.

from pydantic_ai import Agent
from pydantic_ai.models.mistral import MistralModel
from pydantic_ai.providers.mistral import MistralProvider
from pydantic import BaseModel
import logfire
import os
from dotenv import load_dotenv

load_dotenv()


class Cityname(BaseModel):
    city: str
    description: str


logfire.configure(send_to_logfire=os.getenv("LOGFIRE_TOKEN"))
model = MistralModel(
    model_name="mistral-large-latest",
    provider=MistralProvider(api_key=os.getenv("MISTRAL_API_KEY")),
)
agent = Agent(model=model, retries=5, instrument=True, result_type=Cityname)

result_sync = agent.run_sync("Столица Гондураса?")
print(result_sync.data)

Поскольку в проде вы скорее всего должны гонять json-ы, иметь такую штуку и быстро её реализовать очень удобно и приятно.

Weather Agent
Напишем агента для информирования о погодных условиях.
Здесь мы можем начать с модели которую хотели бы получить на выходе.
Например, я хотел бы знать какая скорость ветра, температура и стоит ли мне одеться потеплее или и в шортах норм будет.

class WeatherModel(BaseModel):
    temperature: float
    wind_speed: float
    description: str

Готово. Теперь напишем минимальный системный промпт, чтоб он хоть как то понимал что от него требуется.

system_prompt= """
Ты агент по погоде. Твоя задача предоставить temperature, wind_speed и description
в поле description ты должен дать совет пользователю о том, во что ему одеться думай о комфорте человека в предоставленных тебе погодных условиях
"""

Суммарно наш агент будет выглядеть вот так:

from pydantic_ai import Agent
from pydantic_ai.models.mistral import MistralModel
from pydantic_ai.providers.mistral import MistralProvider
from pydantic import BaseModel
import logfire
import os from dotenv import load_dotenv
load_dotenv()
system_prompt= """
Ты агент по погоде. Твоя задача предоставить temperature, wind_speed и description
в поле description ты должен дать совет пользователю о том, во что ему одеться думай о комфорте человека в предоставленных тебе погодных условиях
"""
class WeatherModel(BaseModel):
    temperature: float
    wind_speed: float
    description: str
    
logfire.configure(send_to_logfire=os.getenv("LOGFIRE_TOKEN"))
model = MistralModel(model_name="mistral-large-latest",                     provider=MistralProvider(api_key=os.getenv("MISTRAL_API_KEY")))
agent = Agent(model=model,               retries=5,               instrument=True,               result_type=WeatherModel,               system_prompt=system_prompt)
result_sync = agent.run_sync('Какая температура сейчас в Москве?')
print(result_sync.data)

Кажется мы что то забыли, но попробуем запустить, вдруг случится магия и все итак заработает.
Конечно же он попытается ответить настолько, насколько может

Агенты в Pydantic AI от вызова LLM до MCP - 2

Но магия не случилась, он ошибся, ведь LLM даже не знает какой сегодня день XD

Агенты в Pydantic AI от вызова LLM до MCP - 3

А теперь добавим тулу Duck Duck Go search.
Аналогичная тула есть и в Langchain, так что принципиально ничего не изменилось.

agent = Agent(model=model, 
              tools=[duckduckgo_search_tool()],
              retries=5, 
              instrument=True, 
              result_type=WeatherModel, 
              system_prompt=system_prompt)
Агенты в Pydantic AI от вызова LLM до MCP - 4

О чудо он плюс минус близок к тому результату который мы от него ожидали. Погода действительно близка к нулю градусам по цельсию и действительно скорость ветра также ближе к нулю.

Travel Assistant

Возьмем задачку с поиском посложнее, сможем ли мы на базе Mistral сделать агента, который сможет подобрать нам рейс?

Снова расписываем модели

class FlightModel(BaseModel):
    flight_company: str = Field(description="Компания которая организует перелет")
    flight_number:str = Field(description="Номер рейса")
    departure_time: datetime = Field(description="Дата и время вылета в локальном времени аэропорта")
    arrival_time: datetime = Field(description="Дата и время прилёта в локальном времени аэропорта")
    price: float = Field(description="Цена за перелет за одного человека")
    price_total: float = Field(description="Цена за перелет суммарно за N человек")
    persons: int = Field(description="количество человек")
    original: str = Field(description="Город вылета , название аэропорта")
    destination: str = Field(description="Город назначения, название аэропорта")
    plane: Optional[str] = Field(description="Название самолета")
    link: str = Field(description="Ссылка на конкретный рейс, который ты предлагаешь. Обязательно должно открыть сам рейс")
    description: str = Field(description="Описание пути c локальным временем вылета, прилета, ценой и остальными параметрами выше")

промптим

flight_prompt= """
Ты агент по поиску рейсов. Ты должен находить ТОЧНЫЕ данные о рейсах.

### Правила:
1. Используй ТОЛЬКО эти сайты: yandextravel.ru, aviasales.ru, ozontravel.ru, tutu.ru
2. Если не находишь рейс - скажи об этом честно
3. Все поля должны быть заполнены ТОЧНЫМИ данными
4. Время указывай в формате: YYYY-MM-DD HH:MM
5. Ссылка должна вести НАПРЯМУЮ на конкретный рейс

### Требуемые данные:
{{  
    "flight_company": str = Field(description="Компания которая организует перелет")
    "flight_number":str = Field(description="Номер рейса")
    "departure_time": datetime = Field(description="Дата и время вылета в локальном времени аэропорта")
    "arrival_time": datetime = Field(description="Дата и время прилёта в локальном времени аэропорта")
    "price": float = Field(description="Цена за перелет за одного человека")
    "price_total": float = Field(description="Цена за перелет суммарно за N человек")
    "persons": int = Field(description="количество человек")
    "original": str = Field(description="Город вылета , название аэропорта")
    "destination": str = Field(description="Город назначения, название аэропорта")
    "plane": Optional[str] = Field(description="Название самолета")
    "link": str = Field(description="Ссылка на конкретный рейс, который ты предлагаешь. Обязательно должно открыть сам рейс")
    "description": str = Field(description="Описание пути c локальным временем вылета, прилета, ценой и остальными параметрами выше")
}}

Исходя из того маршрута, который он тебе предоставил
"""

Вызываем агента на основе моделей Pydantic и промпта, не забываем [7] про search tool

logfire.configure(send_to_logfire=os.getenv("LOGFIRE_TOKEN"))
model = MistralModel(model_name="mistral-large-latest",
                     provider=MistralProvider(api_key=os.getenv("MISTRAL_API_KEY")))
agent = Agent(model=model, 
              tools=[duckduckgo_search_tool()],
              retries=5, 
              instrument=True, 
              result_type=FlightModel, 
              system_prompt=flight_prompt)

result_sync = agent.run_sync('Найди билеты из москвы в тюмень на 24 апреля на два человека после 19:00 по МСК ')
print(result_sync.data)

response:

flight_company='ютэйр' 
flight_number='UT 121' 
departure_time=datetime.datetime(2024, 4, 24, 23, 0) 
arrival_time=datetime.datetime(2024, 4, 25, 4, 40)
price=33009.0 
price_total=66018.0 
persons=2 
original='Москва, Внуково' 
destination='Тюмень, Рощино' 
plane=None 
link='https://avia.tutu.ru/rasp/Moskva--Tyumen/' 
description='Рейс авиакомпании UTair UT 121 вылетает из аэропорта Внуково (VKO) в Москве в 23:00 24 апреля 2024 года по местному времени. Прибытие в аэропорт Рощино (TJM) в Тюмени ожидается в 04:40 25 апреля 2024 года по местному времени. Продолжительность полёта составляет 2 часа 40 минут. Стоимость билета на одного человека составляет 33 009 рублей, а на двух человек - 66 018 рублей. Вы можете забронировать и купить билеты на сайте Туту.ру по ссылке [Авиабилеты Москва - Тюмень]'

Тем временем рейс UT 121

Агенты в Pydantic AI от вызова LLM до MCP - 5

Это конечно неплохо, что он справился с тем выводом, который от него просили, но задачу он так и не выполнил. А рейса Utair из Москвы на сайте на это время в принципе нету. В общем забавный эксперимент едем дальше.

Идем в Google AI Studio [8] и отдаем свои данные на train. Бесплатно, за данные ни о чем, четко, то что нам надо. Так еще и моделька вроде умнее будет…

Реализуем

from pydantic_ai import Agent
from pydantic_ai.models.gemini import GeminiModel
from pydantic_ai.providers.google_gla import GoogleGLAProvider
from pydantic_ai.common_tools.duckduckgo import duckduckgo_search_tool
from pydantic import BaseModel, Field
from datetime import datetime
from typing import Optional
import logfire
import os 
from dotenv import load_dotenv
load_dotenv()

flight_prompt= """
Ты агент по поиску рейсов. Ты должен находить ТОЧНЫЕ данные о рейсах.

### Правила:
1. Используй ТОЛЬКО эти сайты: yandextravel.ru, aviasales.ru, ozontravel.ru, tutu.ru
2. Если не находишь рейс - скажи об этом честно
3. Все поля должны быть заполнены ТОЧНЫМИ данными
4. Время указывай в формате: YYYY-MM-DD HH:MM
5. Ссылка должна вести НАПРЯМУЮ на конкретный рейс

### Требуемые данные:
{{  
    "flight_company": str = Field(description="Компания которая организует перелет")
    "flight_number":str = Field(description="Номер рейса")
    "departure_time": datetime = Field(description="Дата и время вылета в локальном времени аэропорта")
    "arrival_time": datetime = Field(description="Дата и время прилёта в локальном времени аэропорта")
    "price": float = Field(description="Цена за перелет за одного человека")
    "price_total": float = Field(description="Цена за перелет суммарно за N человек")
    "persons": int = Field(description="количество человек")
    "original": str = Field(description="Город вылета , название аэропорта")
    "destination": str = Field(description="Город назначения, название аэропорта")
    "plane": Optional[str] = Field(description="Название самолета")
    "link": str = Field(description="Ссылка на конкретный рейс, который ты предлагаешь. Обязательно должно открыть сам рейс")
    "description": str = Field(description="Описание пути c локальным временем вылета, прилета, ценой и остальными параметрами выше")
}}

Исходя из того маршрута, который он тебе предоставил
"""

class FlightModel(BaseModel):
    flight_company: str = Field(description="Компания которая организует перелет")
    flight_number:str = Field(description="Номер рейса")
    departure_time: datetime = Field(description="Дата и время вылета в локальном времени аэропорта")
    arrival_time: datetime = Field(description="Дата и время прилёта в локальном времени аэропорта")
    price: float = Field(description="Цена за перелет за одного человека")
    price_total: float = Field(description="Цена за перелет суммарно за N человек")
    persons: int = Field(description="количество человек")
    original: str = Field(description="Город вылета , название аэропорта")
    destination: str = Field(description="Город назначения, название аэропорта")
    plane: Optional[str] = Field(description="Название самолета")
    link: str = Field(description="Ссылка на конкретный рейс, который ты предлагаешь. Обязательно должно открыть сам рейс")
    description: str = Field(description="Описание пути c локальным временем вылета, прилета, ценой и остальными параметрами выше")
    
    
logfire.configure(send_to_logfire=os.getenv("LOGFIRE_TOKEN"))
model = GeminiModel(
    'gemini-2.0-flash', provider=GoogleGLAProvider(api_key=os.getenv('GEMINI_API_KEY'))
)
agent = Agent(model=model, 
              tools=[duckduckgo_search_tool()],
              retries=5, 
              instrument=True, 
              result_type=FlightModel, 
              system_prompt=flight_prompt)

result_sync = agent.run_sync('Найди прямой рейс самолета из москвы в тюмень компании Utair на 24 апреля на два человека после 19:00 по МСК ')
print(result_sync.data)

response:

flight_company='Utair' 
flight_number='UT453' 
departure_time=datetime.datetime(2024, 4, 24, 21, 30) 
arrival_time=datetime.datetime(2024, 4, 25, 0, 5) 
price=5929.0 
price_total=11858.0 
persons=2 
original='Москва, Внуково' 
destination='Тюмень, Рощино' 
plane='Boeing 737' 
link='https://avia.tutu.ru/f/Moskva/Tyumen/' 
description='Прямой рейс Utair UT453 из Москвы (Внуково) в Тюмень (Рощино). Вылет 24 апреля в 21:30, прилет 25 апреля в 00:05. Цена за одного человека 5929 рублей, общая стоимость за двоих 11858 рублей.'

Ну чтож такой рейс действительно существует, но по времени и сумме у нас ошибка [9].

Агенты в Pydantic AI от вызова LLM до MCP - 6

В общем то таки, исключая факт того .что цена неверная, как и время, мы наконец смогли найти существующий рейс.

В общем доработав системный промпт, полагаю, можно добиться более качественных результатов, а мы же двинемся дальше.

Agent Delegation

В контексте мультиагентных систем [10] мы рассмотрим один из представленных в документации вариантов реализации в формате Agent Delegation.

Поскольку у нас вроде неплохо получился агент с погодой, возьмем его и добавим еще парочку.

Идейно я хочу получить следующую картину:

Агенты в Pydantic AI от вызова LLM до MCP - 7

Супервизор – анализирует запрос пользователя и решает к какой туле (агенту) ему обратиться.

Currency Agent – агент по курсу валют

Weather Agent – знакомый нам агент по погоде

Joke Agent – агент юморист как в доке Pydantic AI.

На этом моменте в реализации мы отклонимся от подхода Pydantic AI в использовании декораторов, поскольку это выглядит интересно, но как этим пользоваться потом не совсем понятно. Вместо них просто явно пропишем тулы, которыми может пользоваться наш супервизор.

Перейдем к реализации.

Создадим отдельные файлы для схем, промптов и самих агентов, поскольку теперь следит за кодом станет сложнее и проще сделать импорт из файла, чем листать построчно и искать очередную ошибку.

schemas.py
from pydantic import BaseModel
from typing import Optional


class CurrencyModel(BaseModel):
    currency: str
    value: float
    description: str


class WeatherModel(BaseModel):
    temperature: float
    wind_speed: float
    description: str


class JokeModel(BaseModel):
    joke: str


class SupervisorModel(BaseModel):
    weather: Optional[WeatherModel] = None
    currency: Optional[CurrencyModel] = None
    joke: Optional[JokeModel] = None

Соответственно от Currency Model мы хотим получить саму валюту, её значение и какое-то внятное пояснение от агента

От Joke Model просто любая строка сойдет, вообще не паримся.

Weather Model мы уже видели, а вот Supervisor Model лучше знать че она может получить, поэтому перечислим как опциональные модели от предыдуших

Напишем минимальный системный промпт для каждого из агентов, чтобы они понимали свою роль

supervisor_prompt = """
Ты агент супервизор
Твоя задача решить какого агента использовать, в зависмости от намерения пользователя:
- get_currency - агент по валютам, он может сказать сколько стоит одна валюта в другой
- get_weather - агент по погоде 
- generate_joke - агент генератор шуток
Поняв намерение пользователя, вызови тул для выполнения запроса
Верни на выходе результат этого тула
"""

currency_prompt = """
Ты агент для поиска текущего курса валют на основе запроса пользователя
сформируй ответ как:
{{
    currency: str
    value: float
    description: str
}}
currency - USD
value - цена валюты USD

Пример: 
Сколько щас стоит 100 000 RUB в USD
ответ:
{{
    currency: USD
    value: 1 202,21
    description: 100 000 Российских Рублей стоят 1 202,21 USD (доллар США)

}}
"""

weather_prompt = """
Ты агент по погоде. Твоя задача предоставить temperature, wind_speed и description
в поле description ты должен дать совет пользователю о том, во что ему одеться 
думай о комфорте человека в предоставленных тебе погодных условиях
формат вывода:
{{
    temperature: float
    wind_speed: float
    description: str

}}
"""

joke_prompt = """
Сгенерируй шутку
Будь креативным, необычным и нестандартным, ни в чем себя не ограничивая
"""

Ну и теперь переходим к реализации самих агентов, здесь я решил остаться на Gemini, поскольку она не начинает ныть при большем количестве последовательных запросов, в отличие от Mistral.

from pydantic_ai import Agent
from pydantic_ai.models.gemini import GeminiModel
from pydantic_ai.providers.google_gla import GoogleGLAProvider
from pydantic_ai.common_tools.duckduckgo import duckduckgo_search_tool
import logfire
import os
from dotenv import load_dotenv

from prompts import weather_prompt, currency_prompt, joke_prompt, supervisor_prompt
from schemas import WeatherModel, CurrencyModel, JokeModel

load_dotenv()


gemini = GeminiModel(
    model_name="gemini-2.0-flash",
    provider=GoogleGLAProvider(api_key=os.getenv("GEMINI_API_KEY")),
)

logfire.configure(send_to_logfire=os.getenv("LOGFIRE_TOKEN"))


currency_agent = Agent(
    model=gemini,
    tools=[duckduckgo_search_tool()],
    retries=5,
    instrument=True,
    result_type=CurrencyModel,
    system_prompt=currency_prompt,
)

weather_agent = Agent(
    model=gemini,
    tools=[duckduckgo_search_tool()],
    retries=5,
    instrument=True,
    result_type=WeatherModel,
    system_prompt=weather_prompt,
)

joke_agent = Agent(
    model=gemini,
    retries=5,
    instrument=True,
    result_type=JokeModel,
    system_prompt=joke_prompt,
    model_settings={
        "temperature": 0.9,
    },
)


async def get_currency() -> CurrencyModel:
    currency = await currency_agent.run("Пожалуйста, скажи мне курс валют.")
    return currency.data


async def get_weather() -> WeatherModel:
    weather = await weather_agent.run("Какая сейчас погода?")
    return weather.data


async def generate_joke() -> str:
    joke = await joke_agent.run("Расскажи анекдот.")
    return joke.data


supervisor_agent = Agent(
    model=gemini,
    retries=5,
    instrument=True,
    tools=[get_currency, get_weather, generate_joke],
    system_prompt=supervisor_prompt,
)

import asyncio


async def main():
    result = await supervisor_agent.run("Какая температура сейчас в Москве?")
    print(result.data)


asyncio.run(main())

#1
#Message: Какая температура сейчас в Москве?
#Response: В Москве сейчас 14 градусов Цельсия, ожидается облачная с прояснениями погода, возможны осадки, ветер 7.2 метра в секунду.
#2
# Message: Сгенерируй самую странную шутку, что ты когда либо видел
# Response:
#Конечно! Вот для вас необычная шутка:
#Почему хомяки никогда не работают в колл-центрах?
#Потому что они всегда хотят только спать на работе и никак не могут перестать вертеть колесо, даже по телефону! 🐹📞✨
#3 
# Message: Сколько стоит 50 000 рублей в USD?
#К сожалению, я не могу рассчитать стоимость 50 000 рублей в USD, но могу предоставить информацию о курсе 1 USD к EUR. 1 USD стоит 0.93 EUR.

А ну и да, лучше этот код написать в асинхронном стиле, иначе вы рискуете поймать ошибку с event loop. run sync здесь может заработать, но не так радужно как обычный асинхронный run

В общем со своей задачей справились агенты по погоде и по шуткам. Агент по Валюте решил что сойдет просто дать текущий курс, ну да ладно.

В общем и целом это кое-как работает, также как и с агентом по поиску билетов на самолет нужно вносить много правок в системный промпт и дорабатывать его.

MCP

Перейдем к обещанному в начале статьи и самому хайповому варианту – MCP протоколу. На хабре уже есть статьи, описывающие его, поэтому я просто оставлю ссылку здесь [11].

Для запуска MCP и LLM на нужно 2 вещи:

1) Сервер, возьмем из документации [12] mcp-server-fetch [13]

ставим через

uv add mcp-server-fetch

2) Клиент, его пишем сами через Pydantic AI

Здесь я бы хотел реализовать агента, который будет ходить на сайты по блокчейну, их я просто передам через системный промпт.

Вдохновлялся я этим видосом про MCP [14]

Кстати, про defillama.com [15], не советую просить агента ходить туда, поскольку может случиться ситуация. когда он очень много раз стучался и не достучался, а еще потратил много токенов: >1 000 000 input токенов

Агенты в Pydantic AI от вызова LLM до MCP - 8
uv add "pydantic-ai-slim[mcp]"
from pydantic_ai import Agent
from pydantic_ai.models.gemini import GeminiModel
from pydantic_ai.providers.google_gla import GoogleGLAProvider
from pydantic_ai.mcp import MCPServerStdio
from pydantic_ai.mcp import MCPServerHTTP
from pydantic import BaseModel
import logfire
import os
from dotenv import load_dotenv

load_dotenv()


class ChainModel(BaseModel):
    Reasoning: str
    Act: str
    tool_call: str
    Response: str


system_prompt = """
Ты ассистент по блокчейну
Твоя задача анализировать и выдавать инсайты пользователю
Будь полезным и постарайся как можно лучше и эффективнее помочь ему
Твой ответ должен содержать только реальные факты полученные по ссылкам ниже:
https://coinmarketcap.com
https://www.blockchain.com
НЕ ЛЕНИСЬ

пример 1: 
"Human": какой топ 10 протоколов в defi llama
{{"Reasoning": мне необходимо проанализировать api defillama, 
у меня есть к ней доступ, необходимо сходит туда и сделать запрос и вернуть пользователю топ 10 протоколов
"Act": SEARCH [поиск топ 10 протоколов в defi llama]
"tool call": 200 ok, tool is called successfully
"Response": я собрал топ 10 протоколов а также предлагаю следующуие стратегии: ...
}}

Пойми что от тебя требуется, сделай это, верни response
Если у тебя не получилось выполнить запрос и он упал с ошибкой, сообщи это пользователю
"""

logfire.configure(send_to_logfire=os.getenv("LOGFIRE_TOKEN"))

fetch_server = MCPServerStdio(command="uv", args=["run", "-m", "mcp_server_fetch"])
model = GeminiModel(
    "gemini-2.0-flash", provider=GoogleGLAProvider(api_key=os.getenv("GEMINI_API_KEY"))
)
agent = Agent(
    model=model,
    retries=5,
    mcp_servers=[fetch_server],
    system_prompt=system_prompt,
    instrument=True,
    result_type=ChainModel,
)


async def main():
    async with agent.run_mcp_servers():
        result = await agent.run("Привет, отвечай пж по русски всегда")
        while True:
            print(f"n{result.data}")
            user_input = str(input("HumanMessage:"))
            result = await agent.run(
                user_prompt=user_input,
                message_history=result.new_messages(),
            )


if __name__ == "__main__":
    import asyncio

    asyncio.run(main())

Ну и все, теперь запускаем.

По сути изменилось то, что вместо tool мы пишем mcp_servers, а сам сервер запускаем внутри этого скрипта.

Запустив скрипт, мы сможем взаимодействовать с агентом и что-то у него спрашивать

User='Какие сейчас топ токены в крипте? покажи 5 самых дорогих'

Response: 
Reasoning='Я получил данные с coinmarketcap.com. Теперь мне нужно проанализировать HTML контент и выдать топ 5 самых дорогих криптовалют основываясь на их цене.' 
  Act='Топ 5 криптовалют' 
  tool_call='default_api.fetch(url = "https://coinmarketcap.com")' 
  Response='Топ 5 самых дорогих криптовалют на данный момент:nn1.  
  "Bitcoin (BTC): $82,217.58n2.  
  "Ethereum (ETH): $1,560.86n3.  
  "Tether (USDT): $0.9994n4.  
  "XRP (XRP): $2.01n5.  
  "BNB (BNB): $581.44 

Ну и в принципе обмана я не вижу, циферки немного отличаются, но очень-очень близко, учитываем факт того что сайт обновляется постоянно, вместе с ценами.

Агенты в Pydantic AI от вызова LLM до MCP - 9

Summary

В общем и целом, в рамках этой статьи я успел рассказать вам про то, как можно сделать агентов на Pydantic AI, лично мне данный фреймворк понравился. Видно что он еще в разработке и фичи будут добавляться, однако текущие возможности этого фреймворка позволят быстро и без боли [16] сделать вам агентов без лишних зависимостей, без необходимости вникать в графы и прочие вещи, связанные с аналогичными фреймворками.
Про RAG тут пока сомнительно и не понятно. В доке есть пример [17], но выглядит он уже более серьезно, чем RAG на том же лангчейне. Ну и тут есть обсуждение на github [18] как раз по этой тематике.

Автор: Bogdan_m01

Источник [19]


Сайт-источник BrainTools: https://www.braintools.ru

Путь до страницы источника: https://www.braintools.ru/article/14177

URLs in this post:

[1] uv: https://docs.astral.sh/uv/

[2] github: https://github.com/bogdan01m/start-pydantic-ai

[3] Mistral: https://mistral.ai/

[4] зрения: http://www.braintools.ru/article/6238

[5] документации: https://ai.pydantic.dev/agents/#introduction

[6] Pydantic Logfire: https://pydantic.dev/logfire

[7] забываем: http://www.braintools.ru/article/333

[8] Google AI Studio: https://aistudio.google.com/prompts/new_chat

[9] ошибка: http://www.braintools.ru/article/4192

[10] мультиагентных систем: https://ai.pydantic.dev/multi-agent-applications/

[11] ссылку здесь: https://habr.com/ru/articles/879970/

[12] документации: https://modelcontextprotocol.io/examples

[13] mcp-server-fetch: https://github.com/modelcontextprotocol/servers/tree/main/src/fetch

[14] видосом про MCP: https://www.youtube.com/watch?v=OZIWvqcP6so&t=35s

[15] defillama.com: https://defillama.com

[16] боли: http://www.braintools.ru/article/9901

[17] пример: https://ai.pydantic.dev/examples/rag/#example-code

[18] github: https://github.com/pydantic/pydantic-ai/issues/58

[19] Источник: https://habr.com/ru/articles/900270/?utm_source=habrahabr&utm_medium=rss&utm_campaign=900270

www.BrainTools.ru

Рейтинг@Mail.ru
Rambler's Top100