Привет, Хабр!
В последние годы все чаще dстали появляться системы RAG(Retrieval Augmented Generation или "генерация с дополненной выборкой").
Их применяют в областях, где необходима работа со специализированной информацией/документацией и высокая точность генерации с минимальным количеством фактических ошибок.
Возможно, вы уже пользовались такими системы, когда обращались в службу клиентской поддержки или юридические/медицинские организации.
В одной статье сложно охватить все аспекты RAG, поэтому в первой части я расскажу про LangGraph и разберу ключевые концепции. В следующих частях мы перейдем к созданию простых RAG-систем и изучению их основных компонентов.
Установим необходимые библиотеки:
pip install langgraph langchain langchain_core langchain_communityКонцепция
Если сильно упростить, основная идея заключается в следующем: разделить этапы обработки и выборки данных на узлы и определить связи между ними. Такая система называется графом.
Отсюда вытекает важная особенность: у любого графа должно быть состояние (State), которое хранит необходимую информацию. State служит входной схемой для всех узлов графа.
Начинаем начинать
Создадим простой последовательный граф из 3-х узлов, результатом которого будет предложение: 'Я люблю Москву!'.
Для начала создадим состояние с помощью TypedDict, которое будет содержать одно поле: graph_state. Помимо TypedDict можно использовать Pydantic, dataclass.
Пример кода:
from typing_extensions import TypedDict
class State(TypedDict):
graph_state: strОпределим узлы:
def node_1(state):
print("---Node 1---")
return {"graph_state": state['graph_state'] +"Я"}
def node_2(state):
print("---Node 2---")
return {"graph_state": state['graph_state'] +"люблю"}
def node_3(state):
print("---Node 3---")
return {"graph_state": state['graph_state'] +"Москву!"}Пояснение: в качестве аргумента узлы принимают состояние, которое мы определили выше (но узлы пока об этом не знают). В качестве результата каждый узел возвращает обновленное значение поля graph_state.
Самая сложная часть выполнена, осталось добавить узлы в граф и определить связи между ними:
from langgraph.graph import StateGraph, START, END
builder = StateGraph(State)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)Добавили узлы, добавляем ребра:
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_2", "node_3")
builder.add_edge("node_3", END)Здесь у нас появились два не известных узла- START, END. Они определяют вход и выход из нашего графа. Таким образом node1 - стартовый узел, node3 - конечный.
Последний штрих:
graph = builder.compile()
# дополнительно. Показываем схему
display(Image(graph.get_graph().draw_mermaid_png()))
Так как граф является обьектом LangChain, то он имеет метод invoke для вызова.
graph.invoke({"graph_state": ""})---Node 1---
---Node 2---
---Node 3---
{'graph_state': 'Я люблю Москву'}Мы получили требуемый результат, значит можем идти дальше.
Условные ребра
Небольшая, но очень важная тема - условные ребра. Они позволяют динамически выбирать следующий узел графа, в зависимости от ситуации.
Для этого примера используем узлы из предыдущего и добавим несколько новых.
Добавим узел, который вместо "Москву" будет добавлять "Питер".
def node_4(state):
print("---Node 4---")
return {"graph_state": state['graph_state'] +"Питер"}Для выбора следующего узла необходимо создать функцию, которая будет возвращать название следующего узла:
import random
from typing import Literal
def decide_mood(state) -> Literal["node_3", "node_4"]:
user_input = state['graph_state']
if random.random() < 0.5:
return "node_3"
return "node_4"Снова добавим узлы и связи:
from langgraph.graph import StateGraph, START, END
# Build graph
builder = StateGraph(State)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)
builder.add_node("node_4", node_4)builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_conditional_edges("node_2", decide_mood)
builder.add_edge("node_3", END)
builder.add_edge("node_4", END)У нас появилось условное ребро, которое создается с помощью add_conditional_edges, в котором мы указываем: предыдущий узел, функцию, которая отвечает за выбор следующего.
Компилируем граф:
graph = builder.compile()
# View
display(Image(graph.get_graph().draw_mermaid_png()))
---Node 1---
---Node 2---
---Node 4--- #node 3 не вызывался
{'graph_state': 'Я люблю Питер'}Вызов инструментов
Возможно, вы захотите использовать вызов функций в своем RAG, поэтому я решил привести простой пример их использования.
Для начала создадим несколько простых инструментов:
from langchain_core.tools import tool
@tool
def multiply(a: int, b: int) -> int:
"""Multiply a and b.
Args:
a: first int
b: second int
"""
return a * b
@tool
# This will be a tool
def add(a: int, b: int) -> int:
"""Adds a and b.
Args:
a: first int
b: second int
"""
return a + b
@tool
def divide(a: int, b: int) -> float:
"""Divide a and b.
Args:
a: first int
b: second int
"""
return a / b
tools = [add, multiply, divide]Подключим языковую модель, которая поддерживает вызов функций (tool calling):
from langchain_gigachat import GigaChat
llm = GigaChat(verify_ssl_certs=False, credentials="api key", model="GigaChat-2")Я буду использовать GigaChat, для этого потребуется установить:
pip install langchain-gigachatДобавим инстументы к модели:
llm_with_tools = llm.bind_tools(tools)И проверим работоспособность:
llm_with_tools.invoke("what is 2 + 2?")Немного контекста для тех, кто не сталкивался с вызовом инструментов.
Пример вызова модели с вопросом, который требует вызова инструмента, получение результата и передача в модель:
query = "What is 2 * 2?"
messages = [HumanMessage(query)]
ai_msg = llm_with_tools.invoke(messages)
print(ai_msg.tool_calls)
messages.append(ai_msg)
print(messages)В результате получили AIMessage с вызовом функции инструмента, теперь необходимо его вызвать и передать выделенные аргументы:
for tool_call in ai_msg.tool_calls:
selected_tool = {"add": add, "multiply": multiply, "divide": divide}[tool_call["name"].lower()]
tool_msg = selected_tool.invoke(tool_call)
messages.append(tool_msg)Вызов модели с результатом вызова инструмента (передаем все сообщения)
llm_with_tools.invoke(messages)AIMessage(content='Произведение чисел $2$ и $2$ равно $4$.'....Создадим небольшой граф с возможностью вызова функций
Для начала определим состояние:
from typing import Annotated
from langgraph.graph.message import add_messages
class MessagesState(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]Когда граф работает, мы хотим добавлять сообщения в наше состояние, а не перезаписывать (такой вариант определяется по умолчанию). Для этого будем использовать add_messages. Это гарантирует, что любые сообщения добавляются к существующему списку сообщений.
У нас будет всего два узла: узел с вызовом модели и узел с вызовом инструмента.
Первый узел:
def chatbot(state: State):
return {"messages": [llm_with_tools.invoke(state["messages"])]}Добавляет ответ модели, который может содержать вызов инструмента.
Теперь добавим узел, в котором будет происходить вызов инструментов, если это требуется. Мы будем проверять, нужен ли вызов инструмента в последнем сообщении состояния.
import json
from langchain_core.messages import ToolMessage
class ToolNode:
def __init__(self, tools: list) -> None:
self.tools_by_name = {tool.name:tool for tool in tools}
def __call__(self,inputs: dict):
if messages := inputs.get("messages", []):
mesasage = messages[-1] #берем последнее сообщение
else:
raise ValueError("No messages")
outputs = []
for tool_call in mesasage.tool_calls:
tool_res = self.tools_by_name[tool_call["name"]].invoke(tool_call["args"])
outputs.append(ToolMessage(content = json.dumps(tool_res), name=tool_call["name"], tool_call_id=tool_call["id"]))
return {"messages": outputs}Думаю, здесь требуется комментарий: для начала создадим словарь для доступа к нашим инструментам. Выглядит он примерно так:
{'multiply': StructuredTool(name='multiply', description='Multiply a and b.\n\nArgs:\n a: first int\n b: second int', args_schema=<class 'langchain_core.utils.pydantic.multiply'>, func=<function multiply at 0x00000170638BB600>)}
....При вызове узла предполагается, что в состоянии уже есть сообщения: сообщение пользователя и какой то ответ модели. Нас интересует ответ модели, а именно: содержит ли он вызов инструмента. Если вызов инструмента есть, то получаем инструмент по имени и вызываем с извлеченными аргументами. После получения результата добавляем ответ в виде ToolMessage в наше состояние
Остался последний, но не менее важный шаг: написать маршрутизатор, который определяет, когда нужно заканчивать работу.
def route_tools(state: State):
if isinstance(state, list):
ai_message = state[-1]
elif messages := state.get("messages", []):
ai_message = messages[-1]
else:
raise ValueError("No messages")
if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
return "tools"
return ENDОбьяснение: для начала нам нужно получить последнее сообщение в состоянии. После чего проверить, содержит ли оно вызов инструмента. Если последнее сообщение содержит вызов инструмента, то мы переходим в узел "tools", иначе заканчиваем выполнение.
Определяем узлы и связи:
graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
tool_node = ToolNode(tools)
graph_builder.add_node("tools", tool_node)Добавляем ребро с условием
graph_builder.add_conditional_edges(
"chatbot",
route_tools,
{"tools": "tools", END: END}
)graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")
graph = graph_builder.compile()Результат выполнения:
{'messages':
[HumanMessage(content='What is 2 + 2?',
AIMessage(content='', additional_kwargs={'function_call': {'name': 'add', 'arguments': {'a': 2, 'b': 2}},
ToolMessage(content='4', name='add',
AIMessage(content='Сумма двух и двух равна четырем.',
]}На этом я завершаю первую часть. Теперь, когда мы разобрали суть LangGraph и основные принципы его работы, в следующей статье перейдем к изучению ключевых компонентов RAG-систем и рассмотрим их практическое применение.
Подписывайтесь на мой Telegram-канал:
t.me/Viacheslav_Talks
где я регулярно публикую материалы о языковых моделях и искусственном интеллекте, а также анонсирую новые статьи. Подписывайтесь, чтобы не пропустить новые части
Спасибо за прочтение!
