Создание AI-агента на LangChain и OpenRouter

Глава 3: Практические реализации AI-агента: от простого чат-бота до автономного исследователя

В этой главе мы наконец отойдем от теории и погрузимся в написание реального кода. Мы последовательно построим три AI-агента разной сложности, используя OpenRouter и LangChain. Каждый следующий проект будет усложнять предыдущий, добавляя новые инструменты и возможности, иллюстрируя, как из простого набора вызовов API вырастает настоящий автономный инструмент. Наша цель – не просто создать работающий пример, но и понять архитектурные решения, почему мы выбираем те или иные компоненты и как с ними правильно работать.

Проект 1: Базовый агент с планированием и экосистемой инструментов

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

Глава 4: Создание собственных инструментов для агента

В предыдущих главах мы научились создавать цепочки, работать с моделями через OpenRouter и реализовывать базовых агентов, способных использовать предопределенные инструменты. Однако истинная мощь AI-агентов раскрывается только тогда, когда они могут выполнять действия, специфичные для вашего бизнеса или проекта. Стандартные инструменты, такие как поиск в интернете или вычисления, покрывают лишь общие случаи. Чтобы агент стал по-настоящему полезным, ему нужны инструменты, которые знают о вашей базе данных, умеют управлять вашим CRM, развертывать код или анализировать внутренние логи. В этой главе мы подробно разберем, как создавать собственные инструменты (Custom Tools) на Python с использованием LangChain и интегрировать их в агенты, работающие через OpenRouter.

Почему Custom Tools необходимы?

Любой LLM (Large Language Model), работающий через OpenRouter, по своей сути является предсказателем следующего токена. Он не выполняет код и не имеет доступа к внешней среде по умолчанию. Инструменты выступают в качестве "рук" и "ног" модели, позволяя ей взаимодействовать с реальным миром.

Преимущества создания собственных инструментов:

1. Специфичность: Вы можете создать инструмент для работы с уникальным API вашей компании или специфичным файловым форматом.

2. Безопасность: Вы контролируете, какие действия может выполнить агент, ограничивая доступ через API-ключи и логику валидации.

3. Консистентность: Вместо того чтобы заставлять модель генерировать сложные структуры данных (JSON, XML), вы создаете функцию, которая принимает понятные аргументы и возвращает структурированный ответ.

4. Производительность: Инструменты могут кэшировать результаты, выполнять тяжелые вычисления или распараллеливать запросы, недоступные для одной LLM.

Базовая структура инструмента в LangChain

В LangChain инструмент – это класс, который наследуется от базового класса `Tool`. Самый простой способ создать инструмент – использовать декоратор `@tool`. Однако для глубокой интеграции и контроля лучше понимать структуру класса.

Основные компоненты:

– `name`: Уникальное имя инструмента, которое будет видеть LLM.

– `description`: Описание того, что делает инструмент. Это критически важно, так как LLM использует это описание для выбора нужного инструмента (реакция инструмента).

– `args_schema`: Pydantic-модель, описывающая входные параметры. Это помогает LLM понять, какие аргументы нужно сгенерировать.

– `func`: Асинхронная или синхронная функция, которая выполняет логику инструмента.

Разработка первого инструмента: Простой калькулятор

Давайте начнем с простого примера, чтобы понять механику. Представим, что нам нужен калькулятор, который умеет складывать числа. LLM часто ошибаются в арифметике, поэтому передача вычислений наружу – классический сценарий.

Вариант 1: Использование декоратора `@tool`

```python

from langchain_core.tools import tool

import math

@tool

def calculate_factorial(n: int) -> int:

"""Вычисляет факториал числа n. Используйте для операций с множествами и вероятностями."""

return math.factorial(n)

```

В этом примере LangChain автоматически извлечет имя `calculate_factorial`, описание из строки документации и схему аргументов.

Но для серьезных задач нам нужен полный контроль через класс.

Вариант 2: Кастомный класс инструмента

```python

from langchain_core.tools import BaseTool

from typing import Optional, Type

from pydantic import BaseModel, Field

class CalculatorInput(BaseModel):

"""Схема ввода для калькулятора."""

operation: str = Field(description="Операция: сложение (add), вычитание (sub), умножение (mul) или деление (div)")

a: float = Field(description="Первое число")

b: float = Field(description="Второе число")

class CalculatorTool(BaseTool):

name: str = "calculator"

description: str = "Инструмент для выполнения математических операций. Используйте его для любых вычислений."

args_schema: Type[BaseModel] = CalculatorInput

def _run(self, operation: str, a: float, b: float) -> str:

# Синхронная реализация (для простоты, но в продакшене лучше async)

try:

if operation == "add":

return str(a + b)

elif operation == "sub":

return str(a – b)

elif operation == "mul":

return str(a * b)

elif operation == "div":

if b == 0:

return "Ошибка: деление на ноль"

return str(a / b)

else:

return "Ошибка: неизвестная операция"

except Exception as e:

return f"Ошибка вычисления: {e}"

# Инициализация

calc_tool = CalculatorTool()

```

Мы создали строго типизированный инструмент. Обратите внимание на `args_schema`. Когда агент получает доступ к этому инструменту, он знает, что нужно передать `operation`, `a` и `b`. Это снижает галлюцинации модели.

Интеграция через OpenRouter и LangChain Agents

Теперь соберем агента, который использует наш калькулятор. Мы будем использовать `create_react_agent` (или `create_tool_calling_agent` в новых версиях LangChain) и подключим модель через OpenRouter.

Предположим, у нас есть промпт:

"Посчитай (5 + 7) * 3 и затем возведи результат в квадрат".

Агент должен разбить задачу: сначала сложить 5+7, потом умножить на 3, потом возвести в квадрат. Или, если модель умная, она попросит калькулятор сделать 12*3 и затем 36^2.

```python

from langchain_openrouter import ChatOpenRouter

from langchain.agents import create_react_agent, AgentExecutor

from langchain_core.prompts import PromptTemplate

import os

# Настройка OpenRouter

# Убедитесь, что у вас установлен пакет: pip install langchain-openrouter

# Для работы через OpenRouter используем стандартный клиент OpenAI, но с другим base_url

from langchain_openai import ChatOpenAI

# Инициализация модели через OpenRouter

# Можно использовать любой доступный через OpenRouter модель, например, Anthropic Claude или GPT-4

llm = ChatOpenAI(

base_url="https://openrouter.ai/api/v1",

api_key=os.getenv("OPENROUTER_API_KEY"),

model="anthropic/claude-3.5-sonnet", # Пример модели

temperature=0.0

)

# Создаем список инструментов

tools = [CalculatorTool()]

# Шаблон промпта для ReAct агента

prompt_template = """

Отвечай на русском языке. Ты полезный агент, который может использовать инструменты.

Инструменты доступны: {tool_names}

Инструкции:

1. Проверь, нужно ли использовать инструмент для ответа на вопрос.

2. Если да, сгенерируй вызов инструмента в формате:

Thought: [Твои рассуждения]

Action: [Имя инструмента]

Action Input: [Входные данные]

Observation: [Результат выполнения]

3. Повторяй шаги 1-3, пока не получишь окончательный ответ.

4. Не придумывай результаты инструментов.

Вопрос: {input}

{agent_scratchpad}

"""

prompt = PromptTemplate.from_template(prompt_template)

# Создаем агента

agent = create_react_agent(llm, tools, prompt)

# Создаем исполнитель агента

agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# Запуск

question = "Посчитай (5 + 7) * 3 и затем возведи результат в квадрат."

response = agent_executor.invoke({"input": question})

print(response['output'])

```

В этом коде `create_react_agent` подготавливает промпт, вставляя описание инструментов. LLM через OpenRouter получает вопрос, планирует использование инструмента и генерирует структуру вызова. LangChain парсит этот вызов, выполняет `_run` нашего `CalculatorTool` и возвращает результат обратно модели для продолжения рассуждения.

Создание асинхронных инструментов для производительности

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

Для этого нужно:

1. Унаследоваться от `BaseTool`.

2. Реализовать метод `_arun` (асинхронный аналог `_run`).

3. Если `_arun` не реализован, LangChain попытается запустить `_run` в `run_in_executor`, но лучше писать нативный асинхронный код.

Пример инструмента для запроса к внешнему API (Асинхронный):

```python

import aiohttp

from langchain_core.tools import BaseTool

from pydantic import BaseModel, Field

class CryptoPriceInput(BaseModel):

coin_id: str = Field(description="ID криптовалюты (например, bitcoin, ethereum)")

class CryptoPriceTool(BaseTool):

name: str = "get_crypto_price"

description: str = "Получает текущую цену криптовалюты в USD. Используй для финансовых вопросов."

args_schema: Type[BaseModel] = CryptoPriceInput

async def _arun(self, coin_id: str) -> str:

# Асинхронный запрос к CoinGecko API (или любому другому)

url = f"https://api.coingecko.com/api/v3/simple/price?ids={coin_id}&vs_currencies=usd"

try:

async with aiohttp.ClientSession() as session:

async with session.get(url) as response:

if response.status == 200:

data = await response.json()

if coin_id in data:

price = data[coin_id]['usd']

return f"Цена {coin_id} сейчас: ${price}"

return "Монета не найдена"

return f"Ошибка API: {response.status}"

except Exception as e:

return str(e)

def _run(self, coin_id: str) -> str:

# Заглушка для синхронного режима (или можно вызвать синхронный requests)

# В реальном проекте лучше вызывать асинхронный код через asyncio.run

import asyncio

return asyncio.run(self._arun(coin_id))

crypto_tool = CryptoPriceTool()

```

Обратите внимание, что в `_arun` мы используем `aiohttp`. Теперь, если агент будет запущен в асинхронном режиме (например, внутри FastAPI), этот инструмент не будет блокировать поток.

Интеграция с базами данных (SQL Tool)

Один из самых популярных сценариев – позволить агенту общаться с базой данных. В LangChain есть готовые инструменты (`SQLDatabaseToolkit`), но создание своего кастомного SQL-инструмента дает больше контроля над безопасностью.

Допустим, у нас есть SQLite база данных с таблицей `products`.

Нам нужен инструмент, который выполняет запрос и возвращает результат. Чтобы избежать SQL-инъекций, мы не будем передавать сырой SQL от LLM. Вместо этого мы напишем функцию, которая выполняет селект по названию продукта.

```python

from langchain_core.tools import BaseTool

from pydantic import BaseModel, Field

import sqlite3

class ProductSearchInput(BaseModel):

product_name: str = Field(description="Название продукта для поиска")

class ProductSearchTool(BaseTool):

name: str = "sql_product_search"

description: str = "Ищет информацию о продукте в базе данных по его названию. Возвращает цену и описание."

args_schema: Type[BaseModel] = ProductSearchInput

def _run(self, product_name: str) -> str:

# В реальном продакшене используйте пулы соединений (например, SQLAlchemy)

conn = sqlite3.connect('example.db')

cursor = conn.cursor()

Глава 5. Агент с долгосрочной памятью на базе чата

Создание агентов, способных поддерживать непрерывный, осмысленный диалог с пользователем, является одной из ключевых задач при разработке современных интерактивных систем. Обычные чат-боты, работающие в режиме «запрос-ответ», не учитывают предыдущий контекст, что делает взаимодействие фрагментарным и неестественным. Для решения этой проблемы в LangChain реализован механизм, объединяющий под управлением агента компоненты памяти, планирования и выполнения действий. Разделение логики позволяет агенту анализировать историю диалога, извлекать из неё релевантные факты и строить стратегию дальнейшего взаимодействия. Агент с долгосрочной памятью на базе чата (Long-Term Memory Chat Agent) – это архитектура, в которой Large Language Model (LLM) выступает в роли ядра мышления, а специальные хранилища данных (векторные базы или базы ключ-значение) обеспечивают доступ к накопленным знаниям. В этой главе мы детально разберем этапы проектирования, реализации и отладки такого агента с использованием openrouter.ai и LangChain.

Предварительные требования и настройка окружения

Перед началом работы необходимо подготовить инфраструктуру. Поскольку агент будет взаимодействовать с внешними API и базами данных, важно корректно настроить переменные окружения и установить необходимые библиотеки. Основной язык программирования – Python (версия 3.9 и выше). Для работы с LLM через openrouter.ai мы будем использовать стандартные HTTP-клиенты или интеграции LangChain, для работы с памятью – специализированные модули, для хранения эмбеддингов – библиотеку Chromadb или FAISS.

Установка зависимостей:

pip install langchain langchain-openai chromadb faiss-cpu openai python-dotenv

Ключевые компоненты архитектуры

Любой агент в LangChain состоит из трех основных блоков:

1. Модель (LLM): «Мозг» агента. Мы будем использовать модели, доступные через openrouter.ai. Это позволяет выбирать самые мощные модели (например, GPT-4, Claude, Llama) без привязки к одному провайдеру.

2. Память (Memory): Хранилище состояния. Она делится на краткосрочную (контекст текущего окна диалога) и долгосрочную (архив знаний и фактов).

3. Инструменты (Tools): Специальные функции, расширяющие возможности агента (поиск информации, выполнение расчетов, доступ к API).

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

Подключение модели через OpenRouter

OpenRouter предоставляет единый API для множества моделей. Для работы в LangChain нам требуется создать клиент OpenAI-совместимого формата, указав базовый URL OpenRouter.

Пример конфигурации модели:

import os

from langchain_openai import OpenAI

from dotenv import load_dotenv

load_dotenv()

# Указываем URL API OpenRouter вместо стандартного OpenAI

OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY")

BASE_URL = "https://openrouter.ai/api/v1"

llm = OpenAI(

base_url=BASE_URL,

api_key=OPENROUTER_API_KEY,

model="openai/gpt-3.5-turbo", # Можно выбрать любую доступную модель

temperature=0.7

)

Важно помнить, что при использовании агентов с памятью, особенно с длинной историей, стоимость токенов может расти. Поэтому настройка параметров (temperature, max_tokens) критична для оптимизации.

Краткосрочная память как основа диалога

Прежде чем переходить к «архиву» знаний, агент должен помнить текущую беседу. В LangChain для этого используется класс ConversationBufferMemory. Он хранит всю историю сообщений в виде списка, который передается в LLM как часть промпта. Это базовый слой, необходимый для любого чат-агента.

Однако для полноценного агента нам нужно сохранять не только фразы, но и контекст выполнения. Поэтому мы используем ConversationBufferMemory в связке с версией для агентов.

Пример инициализации памяти:

from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)

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

Долгосрочная память: Vector Store и Retrieval

Самая интересная часть – реализация долгосрочной памяти. Хранить весь текст диалога в контексте модели невозможно из-за ограничений длины контекстного окна (context window). Решение – хранить информацию во внешней базе и извлекать её только тогда, когда это необходимо.

Для этого используется механизм Retrieval-Augmented Generation (RAG). Идея следующая:

1. Когда пользователь сообщает важный факт (например, «Я живу в Новосибирске»), агент парсит эту информацию и сохраняет её в векторную базу данных.

2. При последующих вопросах (например, «Какая погода в моем городе?») агент ищет в базе релевантные записи по семантическому сходству (векторный поиск).

3. Найденная информация подставляется в контекст запроса к LLM.

Для реализации нам понадобятся:

Векторная база (Chroma).

Функция вложения (Embeddings). Мы будем использовать встроенные эмбеддинги OpenAI (или доступные через OpenRouter, но обычно для эмбеддингов используют отдельные модели, например, text-embedding-ada-002).

Пример настройки векторного хранилища:

from langchain_community.vectorstores import Chroma

from langchain_openai import OpenAIEmbeddings

# Инициализация эмбеддингов

embeddings = OpenAIEmbeddings(

base_url=BASE_URL,

api_key=OPENROUTER_API_KEY,

model="text-embedding-ada-002" # Эта модель обычно доступна через OpenRouter или OpenAI

)

# Создание или загрузка базы

vectorstore = Chroma(

collection_name="agent_memory",

embedding_function=embeddings,

persist_directory="./chroma_db"

)

Чтобы агент мог использовать эту базу как память, нам нужен специальный инструмент (Tool), который будет выполнять операции сохранения и поиска. Но сначала нужно определить, как именно агент решает, когда сохранять информацию.

Стратегия сохранения информации (Memory Management)

Агент не должен сохранять каждую фразу подряд, иначе база заполнится мусором. Нам нужен механизм принятия решения о важности информации. В LangChain это реализуется через концепцию «Планировщика» (Planner) или нативно через промпт агента.

Мы можем реализовать простую стратегию:

1. Использовать LLM для классификации текущего сообщения пользователя.

2. Если сообщение содержит факты (имя, место, предпочтения), LLM форматирует его в структурированную запись (JSON) и отправляет в инструмент сохранения.

3. Если сообщение – просто реплика в диалоге, оно идет только в краткосрочную память (контекст).

Для реализации этой логики создадим кастомный инструмент (Custom Tool).

Кастомный инструмент долгосрочной памяти

Класс инструмента в LangChain наследуется от BaseTool. Нам нужно два инструмента:

1. `SaveMemory`: сохраняет контекст в Chroma.

2. `LoadMemory`: извлекает контекст по запросу.

Реализация инструмента сохранения:

from langchain.tools import BaseTool

from typing import Optional, Type

from pydantic import BaseModel, Field

class MemoryInput(BaseModel):

content: str = Field(description="Содержание для сохранения в долговременную память")

category: Optional[str] = Field(default="general", description="Категория воспоминания")

class SaveMemoryTool(BaseTool):

name = "save_memory"

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

args_schema: Type[BaseModel] = MemoryInput

def _run(self, content: str, category: str = "general") -> str:

# Сохранение в векторную базу

try:

vectorstore.add_texts(

texts=[content],

metadatas=[{"category": category}]

)

return "Информация успешно сохранена в памяти."

except Exception as e:

return f"Ошибка при сохранении: {e}"

Нам также понадобится инструмент для поиска. Однако, чтобы не перегружать контекст, мы можем сделать «умный» поиск: агент передает запрос, а инструмент возвращает релевантные фрагменты.

class SearchInput(BaseModel):

query: str = Field(description="Поисковый запрос для извлечения воспоминаний")

class LoadMemoryTool(BaseTool):

name = "retrieve_memory"

description = "Используй этот инструмент, чтобы найти сохраненную информацию о пользователе или прошлых событиях, если это необходимо для ответа. Запрос должен содержать ключевые слова или суть того, что вы ищете."

args_schema: Type[BaseModel] = SearchInput

def _run(self, query: str) -> str:

try:

docs = vectorstore.similarity_search(query, k=3)

if not docs:

return "Нет релевантных воспоминаний."

# Форматируем результаты

result_text = "\n".join([doc.page_content for doc in docs])

return f"Найденные воспоминания:\n{result_text}"

except Exception as e:

return f"Ошибка при поиске: {e}"

Сборка агента: Планирование и Очередь

Теперь, когда у нас есть инструменты памяти, краткосрочная память и модель, мы можем собрать агента. Но есть нюанс. Стандартный агент LangChain (AgentExecutor) выполняет инструменты последовательно, пока не получит финальный ответ. Нам же нужна более сложная логика:

1. Анализ текущего ввода.

2. Решение: нужно ли сохранять информацию прямо сейчас? (Это можно делать как отдельный шаг агента, так и через промпт).

3. Решение: нужно ли извлекать информацию из долгосрочной памяти?