Введение
Всем привет, сегодня я расскажу вам о том, как делать можно делать агентов с помощью Pydantic AI.
Pydantic AI – фреймворк от создателей Pydantic – популярной библиотеки для валидации данных в Python с ядром на Rust.
Начнем с простых примеров в виде вызова LLM , а затем постепенно будем усложнять задачу, создавая более сложного агента.
Виртуальное окружение.
В качестве пакетного менеджера в данном проекте используется uv, однако вы можете использовать любой другой, удобный вам, просто имейте это ввиду при установке пакетов.
Репозиторий
Весь код примеров ниже доступен на github
Модель
В качестве стартовой LLM я предлагаю использовать Mistral, а потом переехать на что то более умное, но тоже бесплатное.
Если у вас нет токена, идем на сайт, проходим регистрацию и получаем токен, делаем 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.
Стоит отметить что с точки зрения Pydantic AI агент – это просто интерфейс взаимодействия с LLM, что отвечает в рамках данного фреймворка на вопрос о том, что-же такое этот ваш агент.
Ваш агент может состоять просто из вызова модели, можно добавить к нему системный промпт, формат вывода и другие параметры и это будет все тот же агент, просто с доп функционалом.
А вообще в документации прекрасно описано что такое агент и что он делает, поэтому едем дальше.
Logfire (опционально)
Для трассировки жизненного пути агента, можно воспользоваться Pydantic Logfire. Это позволит нам подробнее посмотреть что делает 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)
Кажется мы что то забыли, но попробуем запустить, вдруг случится магия и все итак заработает.
Конечно же он попытается ответить настолько, насколько может
Но магия не случилась, он ошибся, ведь LLM даже не знает какой сегодня день XD
А теперь добавим тулу Duck Duck Go search.
Аналогичная тула есть и в Langchain, так что принципиально ничего не изменилось.
agent = Agent(model=model,
tools=[duckduckgo_search_tool()],
retries=5,
instrument=True,
result_type=WeatherModel,
system_prompt=system_prompt)
О чудо он плюс минус близок к тому результату который мы от него ожидали. Погода действительно близка к нулю градусам по цельсию и действительно скорость ветра также ближе к нулю.
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 и промпта, не забываем про 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
Это конечно неплохо, что он справился с тем выводом, который от него просили, но задачу он так и не выполнил. А рейса Utair из Москвы на сайте на это время в принципе нету. В общем забавный эксперимент едем дальше.
Идем в Google AI Studio и отдаем свои данные на 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 рублей.'Ну чтож такой рейс действительно существует, но по времени и сумме у нас ошибка.
В общем то таки, исключая факт того .что цена неверная, как и время, мы наконец смогли найти существующий рейс.
В общем доработав системный промпт, полагаю, можно добиться более качественных результатов, а мы же двинемся дальше.
Agent Delegation
В контексте мультиагентных систем мы рассмотрим один из представленных в документации вариантов реализации в формате Agent Delegation.
Поскольку у нас вроде неплохо получился агент с погодой, возьмем его и добавим еще парочку.
Идейно я хочу получить следующую картину:
Супервизор – анализирует запрос пользователя и решает к какой туле (агенту) ему обратиться.
Currency Agent – агент по курсу валют
Weather Agent – знакомый нам агент по погоде
Joke Agent – агент юморист как в доке Pydantic AI.
На этом моменте в реализации мы отклонимся от подхода Pydantic AI в использовании декораторов, поскольку это выглядит интересно, но как этим пользоваться потом не совсем понятно. Вместо них просто явно пропишем тулы, которыми может пользоваться наш супервизор.
Перейдем к реализации.
Создадим отдельные файлы для схем, промптов и самих агентов, поскольку теперь следит за кодом станет сложнее и проще сделать импорт из файла, чем листать построчно и искать очередную ошибку.
schemas.pyfrom 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 протоколу. На хабре уже есть статьи, описывающие его, поэтому я просто оставлю ссылку здесь.
Для запуска MCP и LLM на нужно 2 вещи:
1) Сервер, возьмем из документации mcp-server-fetch
ставим через
uv add mcp-server-fetch2) Клиент, его пишем сами через Pydantic AI
Здесь я бы хотел реализовать агента, который будет ходить на сайты по блокчейну, их я просто передам через системный промпт.
Вдохновлялся я этим видосом про MCP
Кстати, про defillama.com, не советую просить агента ходить туда, поскольку может случиться ситуация. когда он очень много раз стучался и не достучался, а еще потратил много токенов: >1 000 000 input токенов
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
Ну и в принципе обмана я не вижу, циферки немного отличаются, но очень-очень близко, учитываем факт того что сайт обновляется постоянно, вместе с ценами.
Summary
В общем и целом, в рамках этой статьи я успел рассказать вам про то, как можно сделать агентов на Pydantic AI, лично мне данный фреймворк понравился. Видно что он еще в разработке и фичи будут добавляться, однако текущие возможности этого фреймворка позволят быстро и без боли сделать вам агентов без лишних зависимостей, без необходимости вникать в графы и прочие вещи, связанные с аналогичными фреймворками.
Про RAG тут пока сомнительно и не понятно. В доке есть пример, но выглядит он уже более серьезно, чем RAG на том же лангчейне. Ну и тут есть обсуждение на github как раз по этой тематике.
Автор: Bogdan_m01
