Читать книгу Создание AI-агента на LangChain и 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()