Используем языковые модели в AI-агентах. Часть 1. Введение в LangChain

Привет, Хабр!

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

Что такое LangChain?

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

Установка

Для установки необходимо выполнить:

pip install langchain

1. Интерфейс Runnable

Интерфейс Runnable — основа основ для работы со всеми компонентами LangChain. Его реализуют практически все сущности, с которыми нам придется работать.

Основные методы, которые предоставляет интерфейс:

1. invoke/ainvoke: преобразует одиночный входной сигнал в выходной, используется для вызова сущностей, например, языковых моделей.

2. batch/abatch: преобразует множество входных данных в выходные.

3. stream/astream: потоковая передача выходных данных с одного входного сигнала.

ainvoke, abatch, astream — асинхронные вариации.

2. Язык выражений LangChain (LCEL)

Одно из главных преимуществ фреймворка — возможность объединять создаваемые сущности в последовательные «цепочки», где выходные данные одного элемента служат входными данными для следующего.

Давайте разными способами напишем небольшой пример с использованием цепочки:

—RunnableLambda — преобразует вызываемый объект Python в Runnable, который предоставляет преимущества LangChain.

—RunnableSequence — самый важный оператор композиции, поскольку он используется в каждой цепочке и реализует интерфейс Runnable,поэтому для него доступны методы invoke, batch и т.д.

from langchain_core.runnables import RunnableLambda, RunnableSequence  runnable1 = RunnableLambda(lambda x: x + 1) runnable2 = RunnableLambda(lambda x: x + 2)  #создаем цепочку из двух обьект RunnableLambda chain = RunnableSequence(runnable1, runnable2)  #вызывваем цепочку print(chain.invoke(2)) 

Входное значение сначала поступило в runnable1, где было увеличено на 1, а затем поступило в runnable2, где к нему была прибавлена 2. Результатом работы:

5

Этот пример можно написать без использования RunnableSequence, передав входные значения вручную:

langchain_core.runnables import RunnableLambda  runnable1 = RunnableLambda(lambda x: x + 1) runnable2 = RunnableLambda(lambda x: x + 2)  output1 = runnable1.invoke(2) output2 = runnable2.invoke(output1)  print(output2)

Результат работы будет таким же:

5

Еще один вариант, уже с использованием LCEL:

from langchain_core.runnables import RunnableLambda  runnable1 = RunnableLambda(lambda x: x + 1) runnable2 = RunnableLambda(lambda x: x + 2)  chain = runnable1 | runnable2 print(chain.invoke(2))  # 5

Знак | заменяет использование RunnableSequence.

И последний способ реализации, для тех, кому не нравится использование |:

from langchain_core.runnables import RunnableLambda  runnable1 = RunnableLambda(lambda x: x + 1) runnable2 = RunnableLambda(lambda x: x + 2)  chain = runnable1.pipe(runnable2) print(chain.invoke(2))  # 5

Во всех этих примерах мы вызывали цепочки с помощью метода invoke, но мы помним, что помимо invoke нам доступны и другие методы, например, batch:

from langchain_core.runnables import RunnableLambda  runnable1 = RunnableLambda(lambda x: x + 1) runnable2 = RunnableLambda(lambda x: x + 2)  chain = runnable1 | runnable2 print(chain.batch([1, 2, 3]))  # [4, 5, 6]

Каждый элемент был изменен runnable1 и передан в runnable2.

Помимо этого, цепочки можно сделать динамическими в зависимости от входного значения:

from langchain_core.runnables import RunnableLambda  runnable1 = RunnableLambda(lambda x: x + 1) runnable2 = RunnableLambda(lambda x: x + 2)  chain  = RunnableLambda(lambda x: runnable1 if x > 6 else runnable2) chain.invoke(7)

RunnableParallel

Итак, мы рассмотрели четыре способа создания последовательной цепочки, но помимо последовательных существуют и параллельные цепочки. Главное отличие — входные данные передаются не только первому элементу, а сразу всем элементам цепочки.

Для создания такой цепочки необходимо использовать RunnableParallel:

from langchain_core.runnables import RunnableLambda, RunnableParallel  runnable1 = RunnableLambda(lambda x: x + 1) runnable2 = RunnableLambda(lambda x: x + 2)  chain = RunnableParallel({     "runnable_1": runnable1,     "runnable_2": runnable2, })  print(chain.invoke(2))

Результатом работы будет:

{'runnable_1': 3, 'runnable_2': 4}

Как видим, в runnable1 и runnable2 поступили одинаковые входные данные.

RunnableParallel и RunnableSequence можно использовать совместно в одной цепочке.

Создадим словарь, где в качестве значений используем наши RunnableLambda:

mapping = {     "key1": runnable1,     "key2": runnable2, }

Позже этот словарь будет автоматически преобразован в RunnableParallel.

Создадим еще один объект RunnableLambda, который будет складывать результаты выполнения runnable1, runnable2:

runnable3 = RunnableLambda(lambda x: x['key1'] + x['key2'])

Объединяем все в одну цепочку и смотрим на результат:

chain = mapping | runnable3  print(chain.invoke(2)) #7

Из runnable1 на место ‘key1’ вернулась 3, а на место ‘key2’ из runnable2 4.

Function to RunnableLambda

Внутри LCEL функция автоматически преобразуется в RunnableLambda.

Создадим простую функцию и воспользуемся runnable1 из прошлых примеров:

def some_func(x):     return x  chain = some_func | runnable1 print(chain.invoke(2))  # 3

Преобразовать функцию можно явно:

runnable_func = RunnableLambda(some_func)

Использование генератора с помощью stream.

Создадим простой генератор и вспомним, что помимо invoke и batch также имеем stream:

def func(x):     for y in x:         yield str(y)*2          runnable_gen = RunnableLambda(func) for chunk in runnable_gen.stream(range(5)):     print('chunk', chunk)

В результате получим:

chunk 00 chunk 11 chunk 22 chunk 33 chunk 44

Дополнительные методы и RunnablePassthrough

Давайте посмотрим еще на несколько интересных методов, которые предоставляет Runnable.

Допустим, в нашей цепочке есть элемент, который с какой то вероятностью может выполняться некорректно, поэтому мы хотим повторить его выполнение n-е количество раз в надежде на успешное выполнение.

Создадим две функции:

1-я будет просто увеличивать значение входного аргумента на 1

def add_one(x):     return x + 1

2-я будет имитировать некорректную работу:

def bad_function(x):     if random.random() > 0.3:         print('Неудачный вызов')         raise ValueError('bad value')     return x * 2

Объедим их в цепочку и для второй функции воспользуемся методом with_retry, который позволяет повторно вызывать элемент:

chain = RunnableLambda(add_one) | RunnableLambda(bad_function).with_retry(     stop_after_attempt=10, #количество повторений     wait_exponential_jitter=False  # следует ли добавлять задержку между потоврными вызовами )

Возможный результат:

Code failed Code failed Code failed 6

Если спустя 10 попыток не будет достигнуто нужное значение, получим ошибку.

with_fallbacks

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

Для примера возьмем код из примера выше и немного изменим:

def buggy_double(x):     if random.random() > 0.0001: #изменим вероятность         print('Code failed')         raise ValueError('bad value')     return x * 2  #дополнительная функция def failed_func(x):     return x * 2   chain = RunnableLambda(add_one) | RunnableLambda(buggy_double).with_retry(     stop_after_attempt=10,     wait_exponential_jitter=False ).with_fallbacks([RunnableLambda(failed_func)])

После 10 неудачных попыток будет вызвана failed_func.

Bind

bind — метод, который нужен, когда в цепочке необходимо использовать аргумент, которого нет в входных данных или выходных данных предыдущего узла. При этом создается новый объект Runnable.

Создадим простую функцию, результат которой будет отличаться в зависимости от выходных данных:

from langchain_core.runnables import RunnableLambda   def func(main_arg,other_arg = None):     if other_arg:         return {**main_arg, **{"foo": other_arg}}     return main_arg   runnable1 = RunnableLambda(func) bound_runnable1 = runnable1.bind(other_arg="bye") #добавляем аргумент  bound_runnable1.invoke({"bar": "hello"})

В результате получаем:

{'bar': 'hello', 'foo': 'bye'}

RunnablePassthrough

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

Создадим RunnableParallel c использованием RunnablePassthrough:

from langchain_core.runnables import RunnablePassthrough  runnable = RunnableParallel(     origin=RunnablePassthrough(),     modified=lambda x: x + 1 )

И вызовем runnable с помощью batch:

print(runnable.batch([1, 2, 3]))

В результате получим список словарей, которые содержат исходное значение и измененное:

[{'origin': 1, 'modified': 2}, {'origin': 2, 'modified': 3}, {'origin': 3, 'modified': 4}]

Давайте посмотрим на еще один пример. Создадим функцию, которая будет выступать в качестве LLM и возвращать какой то ответ. После получения ответа, будем производить над ним дополнительную обработку:

def fake_llm(prompt):     return {'origin': prompt, 'answer': 'complete'}  chain = RunnableLambda(fake_llm) | {     'orig': RunnablePassthrough(),     "parsed": lambda text: text['answer'][::-1] }  print(chain.invoke('hello'))

Результат работы:

{'orig': {'origin': 'hello', 'answer': 'complete'}, 'parsed': 'etelpmoc'}

Как видим, в «orig» сохранился ответ от fake_llm без изменений, в ‘parsed’ получили новое значение.

assing

Еще один полезный метод, который часто будет использоваться в работе — assing. Он позволяет добавить новое значение в выходной словарь цепочки.

def fake_llm(prompt: str) -> str:     return "complete"   runnable = {'llm1': fake_llm,             'llm2': fake_llm,             } | RunnablePassthrough.assign(     total_chars=lambda inputs: len(inputs['llm1'] + inputs['llm2']) )

В результате мы дополнительно можем получить суммарное количество символов после ответа от наших LLM.

Результат работы:

{'llm1': 'complete', 'llm2': 'complete', 'total_chars': 16}

3. Messages

Messages — основная единица/сущность с которой работает LLM. Они используется для передачи входных и выходных данных, контекста и дополнительной информации. Каждое сообщение имеет свою роль (‘system’, ‘user’, …) и содержание. Перед использованием той или иной роли следует убедиться, что она поддерживается используемой моделью.

Пример сообщения:

("system", "You should only give answers in Spanish.")

Основные роли сообщений:

• system — используется для сообщения модели ее «поведения»

• user — используется для передачи сообщений от пользователя

• assistant — используется для представления ответа модели

• tool — используется для передачи модели результата выполнения инструмента (об этом позже). Поддерживается моделями, которые поддерживают вызов инструментов.

Основные виды сообщений:

• SystemMessage

• HumanMessage

• AIMessage

• ToolMessage

4. Prompt Templates

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

В LangChain существует несколько видов промптов:

• String PromptTemplates — используются для форматирования одной отдельной строки.

• ChatPromptTemplates — используются для форматирования нескольких сообщений.

• MessagesPlaceholder — позволяет вставить список сообщений в определенное место в ChatPromptTemplates.

Все виды реализуют интерфейс Runnable, поэтому поддерживают такие методы, как invoke.

PromptTemplate

Создадим простой пример ,в котором попросим рассказать шутку на какую-то тему:

from langchain_core.prompts import PromptTemplate, FewShotChatMessagePromptTemplate  prompt_template = PromptTemplate.from_template("Tell me a joke about {topic}") print(prompt_template.invoke({"topic": "cats"}))

После вызова получим строку:

Tell me a joke about cats

ChatPromptTemplate

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

from langchain_core.prompts import ChatPromptTemplate  prompt_template = ChatPromptTemplate([     ("system", "You should only give answers in Spanish."),     ("user", "Tell me a joke about {topic}") ])  print(prompt_template.invoke({"topic": "cats"}))

В результате получим список сообщений:

messages=[           SystemMessage(content='You should only give answers in Spanish.', additional_kwargs={}, response_metadata={}),            HumanMessage(content='Tell me a joke about cats', additional_kwargs={}, response_metadata={})          ]

MessagesPlaceholder

Создадим ChatPromptTemplate с одним системным сообщением и местом (MessagesPlaceholder), куда позже могут быть добавлены другие cообщения:

from langchain_core.prompts import MessagesPlaceholder from langchain_core.messages import HumanMessage  prompt_template = ChatPromptTemplate([     ("system", "You should only give answers in Spanish."),     MessagesPlaceholder("msgs") ])  print(prompt_template.invoke({'msgs': [HumanMessage(content='Hi'), HumanMessage(content="Hello")]}))

В результате получим промтп:

messages=[           SystemMessage(content='You must get answers on Spanish', additional_kwargs={}, response_metadata={}),            HumanMessage(content='Hi', additional_kwargs={}, response_metadata={}),           HumanMessage(content='Hello', additional_kwargs={}, response_metadata={})           ]

Второй пример:

MessagesPlaceholder можно составить из нескольких статичных сообщений без использования invoke:

from langchain_core.prompts import MessagesPlaceholder  prompt = MessagesPlaceholder("history") prompt = prompt.format_messages(     history=[         ("system", "You should only give answers in Spanish."),         ("human", "Hello")     ] ) print(prompt)

[  SystemMessage(content='You must get answers on Spanish', additional_kwargs={}, response_metadata={}),   HumanMessage(content='Hello', additional_kwargs={}, response_metadata={}) ]

И последний пример с использованием этого вида подсказок. Создадим шаблон, состоящий из системного сообщения, истории запросов пользователя и ответов модели, а также нового запроса пользователя:

from langchain_core.prompts import MessagesPlaceholder, ChatPromptTemplate  prompt = ChatPromptTemplate.from_messages(     [         ("system", "You should only give answers in Spanish."),         MessagesPlaceholder("history"),         ("human", "{question}")     ] )  print(prompt.invoke(     {         "history": [('human', "what is 5 +2?"), ("ai", "5+2 is 7")],         "question": "now now multiply that by 4"     } ))

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

messages=[SystemMessage(content='You must get answers on Spanish', additional_kwargs={}, response_metadata={}),            HumanMessage(content='what is 5 +2?', additional_kwargs={}, response_metadata={}),           AIMessage(content='5+2 is 7', additional_kwargs={}, response_metadata={}),            HumanMessage(content='now now multiply that by 4', additional_kwargs={}, response_metadata={})          ]

Теперь мы знаем достаточно, чтобы перейти к простому использованию LLM)

5. ChatModels

Наконец то, мы можем познакомиться со способами использования языковых моделей.

Начнем с того, что все модели происходят от базового класса BaseLanguageModel, который реализует интерфейс Runnbale, поэтому мы можем вызывать модели с помощью invoke,batch и т.д.

Чтобы создать модель существует множество способов, например, такой с использование OpenAI:

import getpass import os  if not os.environ.get("OPENAI_API_KEY"):   os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter API key for OpenAI: ")  from langchain_openai import ChatOpenAI  llm = ChatOpenAI(model="gpt-4o-mini")

Но мне больше нравится использование моделей с Hugging Face. Как раз для этого у Hugging Face и LangChain существует партнерский пакет, который предоставляет простой доступ к LLM.

Для этого необходимо установить:

pip install langchain-huggingface

Существует несколько способов использовать модель с Hugging Face:

С помощью HuggingFacePipeline:

from langchain_huggingface import HuggingFacePipeline  llm = HuggingFacePipeline.from_model_id(     model_id=model_repo_id,     task="text-generation",     pipeline_kwargs={         "max_new_tokens": 100,         "top_k": 50,         "temperature": 0.1,     } )

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

С помощью HuggingFaceEndpoint:

llm = HuggingFaceEndpoint(     repo_id=model_repo_id,     temperature=0.8,     task="text-generation",     max_new_tokens=1000,     do_sample=False, )

В этом случае будет использовано serverless API, поэтому необходимо создать аккаунт на HuggingFace и получить huggingface_token.

Про параметры моделей я рассказывал в предыдущей статье.

Пример использования

Рассмотрим простой вариант использования модели и шаблонов подсказок.

from langchain_core.prompts import ChatPromptTemplate  model_repo_id = "meta-llama/Meta-Llama-3-8B-Instruct"  llm = HuggingFaceEndpoint(     repo_id=model_repo_id,     temperature=0.8,     task="text-generation",     max_new_tokens=1000,     do_sample=False, )  #ChatHuggingFace помогает составить правильный запрос к модели и является  #оберткой поверх llm model = ChatHuggingFace(llm=llm)  prompt = ChatPromptTemplate.from_messages([     ("system", "You should only give answers in Spanish."),     ("user", "Hello, how are you?") ])  #используем LCEL chain = prompt | model  print(chain.invoke({})) 

В результате получим ответ модели:

content='Hola, estoy bien, ¿y tú?' additional_kwargs={} response_metadata={'token_usage': ChatCompletionOutputUsage(completion_tokens=10, prompt_tokens=29, total_tokens=39), 'model': '', 'finish_reason': 'stop'} id='run-3c0fe167-c083-4274-9d21-f29d9ed7873a-0'

Заключение

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

Мой телеграмм канал, там я пишу про LLM и выход моих статей:

https://t.me/Viacheslav_Talks