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

В одной из прошлых статей я рассказывал про дообучение языковых моделей, сегодня же я хочу поговорить про практическое использование 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