Привет, Хабр!
В последние годы все чаще 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-канал:
где я регулярно публикую материалы о языковых моделях и искусственном интеллекте, а также анонсирую новые статьи. Подписывайтесь, чтобы не пропустить новые части
Спасибо за прочтение!
