⚡ Разработка AI-агентов в виде графов ⚡

GigaGraph — это библиотека, дающая возможность работать с LLM (большие языковые модели) для создания приложений, которые используют множество взаимодействующих цепочек (акторов) и сохраняют данные о состоянии. Так как в основе GigaGraph лежит GigaChain, предполагается совместное использование обоих библиотек.

Основной сценарий использования GigaGraph — добавление циклов в приложения с LLM. Для этого библиотека добавляет в LangChain Expression Language возможность работать с множеством цепочек на каждой из итераций вычислительного цикла. Использование циклов позволяет реализовать поведение агента, когда приложению нужно многократно вызывать LLM и спрашивать, какое действие нужно выполнить следующим.

Следует отметить, что GigaGraph не предназначена для создания DAG (ориентированного ациклического графа). Для решения этой задачи используйте стандартные возможности LangChain Expression Language.

1. AgenticRAG + GigaChat

Для установки используйте менеджер пакетов pip:

pip install gigagraph

Если вы хотите использовать gigagraph вместе с GigaChat, то рекомендуется сразу уже установить пакет gigachain-community:

pip install gigachain_community

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

Для работы агента потребуется установить некоторые пакеты GigaChain и использовать в качестве демонстрации сервис Tavily:

State in LangGraph can be pretty general, but to keep things simpler to start, we'll show off an example where the graph's state is limited to a list of chat messages using the built-in MessageGraph class. This is convenient when using LangGraph with LangChain chat models because we can directly return chat model output.

First, install the GigaChain OpenAI integration package:

pip install gigachain_openai

We also need to export some environment variables:

export OPENAI_API_KEY=sk-...

And now we're ready! The graph below contains a single node called "oracle" that executes a chat model, then returns the result:

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langgraph.graph import END, MessageGraph

model = ChatOpenAI(temperature=0)

graph = MessageGraph()

graph.add_node("oracle", model)
graph.add_edge("oracle", END)

graph.set_entry_point("oracle")

runnable = graph.compile()

Let's run it!

runnable.invoke(HumanMessage("What is 1 + 1?"))

[HumanMessage(content='What is 1 + 1?'), AIMessage(content='1 + 1 equals 2.')]

So what did we do here? Let's break it down step by step:

1. First, we initialize our model and a MessageGraph.
2. Next, we add a single node to the graph, called "oracle", which simply calls the model with the given input.
3. We add an edge from this "oracle" node to the special string END ("__end__"). This means that execution will end after the current node.
4. We set "oracle" as the entrypoint to the graph.
5. We compile the graph, translating it to low-level pregel operations ensuring that it can be run.

Then, when we execute the graph:

1. LangGraph adds the input message to the internal state, then passes the state to the entrypoint node, "oracle".
2. The "oracle" node executes, invoking the chat model.
3. The chat model returns an AIMessage. LangGraph adds this to the state.
4. Execution progresses to the special END value and outputs the final state.

And as a result, we get a list of two chat messages as output.

As an aside for those already familiar with LangChain - add_node actually takes any function or runnable as input. In the above example, the model is used "as-is", but we could also have passed in a function:

def call_oracle(messages: list):
    return model.invoke(messages)

graph.add_node("oracle", call_oracle)

Just make sure you are mindful of the fact that the input to the runnable is the entire current state. So this will fail:

# This will not work with MessageGraph!
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant named {name} who always speaks in pirate dialect"),
    MessagesPlaceholder(variable_name="messages"),
])

chain = prompt | model

# State is a list of messages, but our chain expects a dict input:
#
# { "name": some_string, "messages": [] }
#
# Therefore, the graph will throw an exception when it executes here.
graph.add_node("oracle", chain)

Now, let's move onto something a little bit less trivial. LLMs struggle with math, so let's allow the LLM to conditionally call a "multiply" node using tool calling.

We'll recreate our graph with an additional "multiply" that will take the result of the most recent message, if it is a tool call, and calculate the result. We'll also bind the calculator's schema to the OpenAI model as a tool to allow the model to optionally use the tool necessary to respond to the current state:

from langchain_core.tools import tool
from langgraph.prebuilt import ToolNode

@tool
def multiply(first_number: int, second_number: int):
    """Multiplies two numbers together."""
    return first_number * second_number

model = ChatOpenAI(temperature=0)
model_with_tools = model.bind_tools([multiply])

builder = MessageGraph()

builder.add_node("oracle", model_with_tools)

tool_node = ToolNode([multiply])
builder.add_node("multiply", tool_node)

builder.add_edge("multiply", END)

builder.set_entry_point("oracle")

Now let's think - what do we want to have happened?

• If the "oracle" node returns a message expecting a tool call, we want to execute the "multiply" node
• If not, we can just end execution

We can achieve this using conditional edges, which call a function on the current state and routes execution to a node the function's output.

Here's what that looks like:

from typing import Literal

def router(state: List[BaseMessage]) -> Literal["multiply", "__end__"]:
    tool_calls = state[-1].additional_kwargs.get("tool_calls", [])
    if len(tool_calls):
        return "multiply"
    else:
        return "__end__"

builder.add_conditional_edges("oracle", router)

If the model output contains a tool call, we move to the "multiply" node. Otherwise, we end execution.

Great! Now all that's left is to compile the graph and try it out. Math-related questions are routed to the calculator tool:

runnable = builder.compile()

runnable.invoke(HumanMessage("What is 123 * 456?"))

[HumanMessage(content='What is 123 * 456?'),
 AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_OPbdlm8Ih1mNOObGf3tMcNgb', 'function': {'arguments': '{"first_number":123,"second_number":456}', 'name': 'multiply'}, 'type': 'function'}]}),
 ToolMessage(content='56088', tool_call_id='call_OPbdlm8Ih1mNOObGf3tMcNgb')]

While conversational responses are outputted directly:

runnable.invoke(HumanMessage("What is your name?"))

[HumanMessage(content='What is your name?'),
 AIMessage(content='My name is Assistant. How can I assist you today?')]

Now, let's go over a more general cyclic example. We will recreate the AgentExecutor class from LangChain. The agent itself will use chat models and tool calling. This agent will represent all its state as a list of messages.

We will need to install some GigaChain packages, as well as Tavily to use as an example tool.

pip install -U gigachain gigachain_openai tavily-python

Также для доступа к OpenAI и Tavily API понадобится задать переменные среды:

export ANTHROPIC_API_KEY=sk-...

При желании вы можете использовать LangSmith:

export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY=lsv2_sk_...

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

from langchain_community.tools.tavily_search import TavilySearchResults

tools = [TavilySearchResults(max_results=1)]

Оберните инструменты в GigaGraph ToolExecutor — класс, который принимает объекты запуска инструмента ToolInvocation, вызывает инструмент и возвращает ответ. Объект ToolInvocation — произвольный класс с атрибутами tool и tool_input.

from langgraph.prebuilt import ToolNode

tool_node = ToolNode(tools)

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

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

from langchain_openai import ChatOpenAI
from langgraph.checkpoint import MemorySaver
from langgraph.graph import END, StateGraph, MessagesState
from langgraph.prebuilt import ToolNode

# Параметр streaming=True включает потоковую передачу токенов
# Подробнее в разделе Потоковая передача.
model = ChatOpenAI(temperature=0, streaming=True)

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

model = model.bind_tools(tools)

Основным графом gigagraph является StatefulGraph. Этот граф параметризован объектом состояния, который он передает каждой вершине. В свою очередь каждая вершина возвращает операции для обновления состояния. Операции могут либо задавать (SET) определенные атрибуты состояния (например, переписывать существующие значения), либо добавлять ()ADD данные к существующим атрибутам. Будет операция задавать или добавлять данные, определяется аннотациями объекта состояния, который используется для создания графа.

В приведенном примере отслеживаемое состояние представлено в виде списка сообщений. Поэтому нужно чтобы каждая вершина добавляла сообщения в список.

Для этого используйте TypedDict с одним ключом (messages) и аннотацией, указывающей на то, что в атрибут messages можно только добавлять данные.

from typing import TypedDict, Annotated

def add_messages(left: list, right: list):
    """Add-don't-overwrite."""
    return left + right

class AgentState(TypedDict):
    # The `add_messages` function within the annotation defines
    # *how* updates should be merged into the state.
    messages: Annotated[list, add_messages]

Теперь нужно определить несколько разных вершин графа. В langgraph вершина может быть представлена в виде функции или исполняемого интерфейса. Для описываемого примера понадобятся две основных вершины:

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

Также нужно определить ребра графа. Часть ребер могут зависеть от условий (условные ребра). Это связанно с тем, что в зависимости от вывода вершины могут быть реализованы различные пути развития событий. При этом неизвестно какой путь будет выбран до момента обращения к вершине. Какой путь выбрать LLM решает самостоятельно.

Разница между обычным и условным ребром графа:

• В случае условного ребра, после вызова агента:

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

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

Определите вершины и функцию, которая будет решать какое из условных ребер выполнять.

model = ChatAnthropic(model="claude-3-5-sonnet-20240620", temperature=0).bind_tools(tools)

def should_continue(state): messages = state['messages'] last_message = messages[-1] # Приложение останавливается, если нет вызова функции if "function_call" not in last_message.additional_kwargs: return "end" # В противном случае выполнение продолжается else: return "continue"

def call_model(state): messages = state['messages'] response = model.invoke(messages) # Возвращается список, который будет добавлен к существующему списку сообщений return {"messages": [response]}

def call_tool(state): messages = state['messages'] # Благодаря условию continue # приложение знает, что последнее сообщение содержит вызов функции last_message = messages[-1] # Создание ToolInvocation из function_call action = ToolInvocation( tool=last_message.additional_kwargs["function_call"]["name"], tool_input=json.loads(last_message.additional_kwargs["function_call"]["arguments"]), ) # Вызов tool_executor и получение ответа response = tool_executor.invoke(action) # Использование ответа для создания сообщения FunctionMessage function_message = FunctionMessage(content=str(response), name=action.tool) # Возвращение списка, который будет добавлен к существующему списку сообщений return {"messages": [function_message]}

### Определите граф

```python
from langgraph.graph import StateGraph, END
# Задайте новый граф
workflow = StateGraph(AgentState)

# Задайте две вершины, которые будут работать в цикле
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)

# Задайте точку входа `agent`
# Точка входа указывает вершину, котора будет вызвана в первую очередь
workflow.set_entry_point("agent")

# Создайте условное ребро
workflow.add_conditional_edges(
    # Определите начальную вершину. В этом примере используется вершина `agent`.
    # Это задает ребра, которые будут использованы после вызова вершины `agent`.
    "agent",
    # Передайте функцию, которая определяет какую вершину вызвать дальше.
    should_continue,
    # Передайте структуру (map), в которой ключами будут строки, а значениями другие вершины.
    # END — зарезервированная вершна, указываящая на то, что граф должен завершиться.
    # После вызова `should_continue` вывод функции сравнивается с ключами в структуре.
    # После чего вызывается соотвествующая выводу вершина.
    {
        # If `tools`, then we call the tool node.
        # Если значение `tools`, вызывается вершина, ответственная за обращение к интрсументам.
        "continue": "action",
        # В противном случае граф заканчивается.
        "end": END
    }
)

# Добавьте обычное ребро, соединяющее вершины `tools` и `agent`.
# Ребро задает путь при котором после вызова вершины `tools`, вызывается вершина `agent`.
workflow.add_edge('action', 'agent')

# Скомпилируйте все предыдущие этапы в исполняемый интерфейс GigaChain.
# Теперь граф можно использовать также, как и другие исполняемые интерфейсы.
app = workflow.compile()

Скомпилированный исполняемый интерфейс принимает на вход список сообщений:

from langchain_core.messages import HumanMessage

inputs = {"messages": [HumanMessage(content="what is the weather in sf")]}
app.invoke(inputs)

Работа интерфейса занимает некоторое время. Чтобы наблюдать за результатом работы в прямом эфире, вы можете включить потоковую передачу.

GigaGraph поддерживает несколько разных способов потоковой передачи.

GigaGraph предоставляет возможность потоковой передачи результата вызова каждой из вершин графа по мере обращения к ним.

inputs = {"messages": [HumanMessage(content="what is the weather in sf")]}
for output in app.stream(inputs):
    # stream() возвращает словари с парами `Вершина графа — вывод`
    for key, value in output.items():
        print(f"Output from node '{key}':")
        print("---")
        print(value)
    print("\n---\n")

"Based on the search results, I can tell you that the current weather in San Francisco is:\n\nTemperature: 60 degrees Fahrenheit\nConditions: Foggy\n\nSan Francisco is known for its microclimates and frequent fog, especially during the summer months. The temperature of 60°F (about 15.5°C) is quite typical for the city, which tends to have mild temperatures year-round. The fog, often referred to as "Karl the Fog" by locals, is a characteristic feature of San Francisco\'s weather, particularly in the mornings and evenings.\n\nIs there anything else you\'d like to know about the weather in San Francisco or any other location?"

Библиотека дает доступ к потоковой передачи токенов модели по мере их возникновения на каждой из вершин. В приведенном примере только вершина agent может возвращать токены модели. Для работы этой функциональность нужно чтобы модель поддерживала работу в режиме потоковой передачи токенов.

final_state = app.invoke(
    {"messages": [HumanMessage(content="what about ny")]},
    config={"configurable": {"thread_id": 42}}
)
final_state["messages"][-1].content

"Based on the search results, I can tell you that the current weather in New York City is:\n\nTemperature: 90 degrees Fahrenheit (approximately 32.2 degrees Celsius)\nConditions: Sunny\n\nThis weather is quite different from what we just saw in San Francisco. New York is experiencing much warmer temperatures right now. Here are a few points to note:\n\n1. The temperature of 90°F is quite hot, typical of summer weather in New York City.\n2. The sunny conditions suggest clear skies, which is great for outdoor activities but also means it might feel even hotter due to direct sunlight.\n3. This kind of weather in New York often comes with high humidity, which can make it feel even warmer than the actual temperature suggests.\n\nIt's interesting to see the stark contrast between San Francisco's mild, foggy weather and New York's hot, sunny conditions. This difference illustrates how varied weather can be across different parts of the United States, even on the same day.\n\nIs there anything else you'd like to know about the weather in New York or any other location?"

Используйте библиотеку если вам нужна поддержка циклов.

Если обычной работы с цепочками для решения ваших задач достаточно, используйте основные возможности LangChain Expression Language.

These guides show how to use LangGraph in particular ways.

If you are running LangGraph in async workflows, you may want to create the nodes to be async by default. For a walkthrough on how to do that, see this documentation

Sometimes language models take a while to respond and you may want to stream tokens to end users. For a guide on how to do this, see this documentation

LangGraph comes with built-in persistence, allowing you to save the state of the graph at point and resume from there. For a walkthrough on how to do that, see this documentation

LangGraph comes with built-in support for human-in-the-loop workflows. This is useful when you want to have a human review the current state before proceeding to a particular node. For a walkthrough on how to do that, see this documentation

Agents you create with LangGraph can be complex. In order to make it easier to understand what is happening under the hood, we've added methods to print out and visualize the graph. This can create both ascii art and pngs. For a walkthrough on how to do that, see this documentation

With "time travel" functionality you can jump to any point in the graph execution, modify the state, and rerun from there. This is useful for both debugging workflows, as well as end user-facing workflows to allow them to correct the state. For a walkthrough on how to do that, see this documentation

Пример приложения-исполнителя принимает на вход список сообщений и также возвращает список сообщений на выходе. Состояние агента также представлено в виде списка сообщений. Представленный пример использует вызов функций OpenAI.

• Getting Started Notebook. Базовый пример, демонстрирующий пошаговое создание приложения исполнителя агентов.
• High Level Entrypoint. Пример демонстрирует как можно использовать высокоуровневую точку входа для исполнителя чат-агента.

Вариации примеров

We also have a lot of examples highlighting how to slightly modify the base chat agent executor. These all build off the getting started notebook so it is recommended you start with that first.

• Human-in-the-loop. Пример демонстрирует как реализовать подход «человек-в-цикле».
• Принудительный вызов инструмента. Пример демонстрирует как всегда вызывать определенный инструмент в первую очередь.
• Ответ в заданном формате. Пример демонстрирует, как принудительно получить ответ агента в заданном формате.
• Динамический вывод результата использования инструмента. Пример демонстрирует, как агент может самостоятельно решать возвращать результат использования инструмента пользователю или нет.
• Управление этапами работы агента. Пример демонстрирует, как можно более детально управлять промежуточными этапами работы агента.

Примеры приложений-исполнителей, использующих агенты LangChain.

• Getting Started Notebook. Базовый пример, демонстрирующий пошаговое создание приложения исполнителя агентов.
• High Level Entrypoint. Пример демонстрирует как можно использовать высокоуровневую точку входа для исполнителя чат-агента.

Вариации примеров

Примеры небольших изменений, которые можно сделать при разработке исполнителя чат-агента. Приведенные вариации основаны на примере Getting Started Notebook.

• Human-in-the-loop. Пример демонстрирует как реализовать подход «человек-в-цикле».
• Принудительный вызов инструмента. Пример демонстрирует как всегда вызывать определенный инструмент в первую очередь.
• Управление этапами работы агента. Пример демонстрирует, как можно более детально управлять промежуточными этапами работы агента.

The following notebooks implement agent architectures prototypical of the "plan-and-execute" style, where an LLM planner decomposes a user request into a program, an executor executes the program, and an LLM synthesizes a response (and/or dynamically replans) based on the program outputs.

• Plan-and-execute: a simple agent with a planner that generates a multi-step task list, an executor that invokes the tools in the plan, and a replanner that responds or generates an updated plan. Based on the Plan-and-solve paper by Wang, et. al.
• Reasoning without Observation: planner generates a task list whose observations are saved as variables. Variables can be used in subsequent tasks to reduce the need for further re-planning. Based on the ReWOO paper by Xu, et. al.
• LLMCompiler: planner generates a DAG of tasks with variable responses. Tasks are streamed and executed eagerly to minimize tool execution runtime. Based on the paper by Kim, et. al.

When output quality is a major concern, it's common to incorporate some combination of self-critique or reflection and external validation to refine your system's outputs. The following examples demonstrate research that implement this type of design.

• Basic Reflection: add a simple "reflect" step in your graph to prompt your system to revise its outputs.
• Reflexion: critique missing and superflous aspects of the agent's response to guide subsequent steps. Based on Reflexion, by Shinn, et. al.
• Giga Reflexion: реализация Reflexion на GigaChat
• Language Agent Tree Search: execute multiple agents in parallel, using reflection and environmental rewards to drive a Monte Carlo Tree Search. Based on LATS, by Zhou, et. al.

• Совместная работа нескольких агентов. Пример демонстрирует как создать двух агентов, которые работают вместе для решения задачи.
• Несколько агентов с «руководителем». Пример демонстрирует как организовать работу агентов используя LLM в роли «руководителя», который решает как распределять работу.
• Иерархичные команды агентов: пример демонстрирует как организовать «команды» агентов, которые будут взаимодействовать для решения задачи, в виде вложенных графов.

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

• Оценка чат-бота с помощью симуляции взаимодействия нескольких агентов.. В примере показано как симулировать диалог «виртуального пользователя» с чат-ботом.

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

• Chat bot evaluation as multi-agent simulation: how to simulate a dialogue between a "virtual user" and your chat bot
• Evaluating over a dataset: benchmark your assistant over a LangSmith dataset, which tasks a simulated customer to red-team your chat bot.

Ответ модели может занимать продолжительное время и вам может потребоваться на лету отображать пользователям результат работы модели. Пример.

GigaGraph позволяет сохранять состояние графа в определенный момент времени и потом возобновлять работу с этого состояния. Пример.

GigaGraph поддерживает процесс, при котором необходимо участие человека, проверяющего текущее состояние графа перед переходом к следующей вершине. Пример такого подхода — в документации.

GigaGraph предоставляет доступ к нескольким новым интерфейсам.

Основная точка входа — класс StateGraph.

from langgraph.graph import StateGraph

Класс ответственный за создание графа. Этот граф параметризован объектом состояния, который он передает каждой вершине.

    def __init__(self, schema: Type[Any]) -> None:

При создании графа нужно передать схему состояния. Каждая вершина будет возращать операции для обновления этого состояния. Операции могут либо задавать (SET) определенные атрибуты состояния (например, переписывать существующие значения), либо добавлять ()ADD данные к существующим атрибутам. Будет операция задавать или добавлять данные, определяется аннотациями объекта состояния, который используется для создания графа.

Схему состояния рекомендуется задавать с помощью типизированного словаря: from typing import TypedDict

После создания схемы вы можете аннотировать атрибуты с помощью from typing imoport Annotated. Сейчас поддерживается только одна аннотация — import operator; operator.add. Аннотация указывает, что каждая вершина, которая возвращает этот атрибут добавляет новые данные к существующему значению.

Пример состояния:

class AgentState(TypedDict):

input: str

agent_outcome: Union[AgentAction, AgentFinish, None]

intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add]

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

```python
# Инициализируйте StateGraph с помощью состояния AgentState
graph = StateGraph(AgentState)
# Создайте вершины и ребра
...
# Скомпилируйте граф
app = graph.compile()

# На вход должен передаваться словарь, так как состояние создано как TypedDict
inputs = {
   # Пример входный данныъ
   "input": "hi"
   # Предположим, что `agent_outcome` задается графом как некоторая точка
   # Передавать значение не нужно, по умолчанию оно будет None
   # Предположим, что граф со временем наполняет `intermediate_steps`
   # Передавать значение не нужно, по умолчанию список будет пустым
   # Список `intermediate_steps` будет представлен в виде пустого списка, а не None потому,
   # что он аннотирован с помощью `operator.add`
}

    def add_node(self, key: str, action: RunnableLike) -> None:

This method adds a node to the graph. Добавляет вершину графа. Принимает два параметра:

• key — Уникальная строка с названием вершины.
• action — действие, которое выполняется при вызове вершины. Выражается в виде функции или исполняемого интерфейса.

    def add_edge(self, start_key: str, end_key: str) -> None:

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

• start_key — строка с названием начальной вершины. Название вершины должно быть зарегистрировано в графе.
• end_key — строка с названием конечной вершины. Название вершины должно быть зарегистрировано в графе.

    def add_conditional_edges(
        self,
        start_key: str,
        condition: Callable[..., str],
        conditional_edge_mapping: Dict[str, str],
    ) -> None:

Создает условное ребро. Позволяет задавать пути развития событий в зависимости от результата вызова начальной вершины. Принимает три параметра:

• start_key — строка с названием начальной вершины. Название вершины должно быть зарегистрировано в графе.
• condition — функция, которая вызывается для определения пути развития событий. На вход принимает результат вызова начальной вершины. Возвращает строку, зарегистрированную в структуре conditional_edge_mapping, которая указывает в соответствии с каким ребром будут развиваться события.
• conditional_edge_mapping — структура (map) строка-строка. В качестве ключа задается название ребра, которое может вернуть condition. В качестве значения задается вершина, которые будет вызваны если condition вернет соответствующее название ребра.

    def set_entry_point(self, key: str) -> None:

Точка входа в граф. Задает вершину, которая будет вызвана в самом начале. Принимает один параметр:

• key — название вершины, которую нужно вызывать в первую очередь.

    def set_conditional_entry_point(
        self,
        condition: Callable[..., str],
        conditional_edge_mapping: Optional[Dict[str, str]] = None,
    ) -> None:

This method adds a conditional entry point. What this means is that when the graph is called, it will call the condition Callable to decide what node to enter into first.

• condition: A function to call to decide what to do next. The input will be the input to the graph. It should return a string that is present in conditional_edge_mapping and represents the edge to take.
• conditional_edge_mapping: A mapping of string to string. The keys should be strings that may be returned by condition. The values should be the downstream node to call if that condition is returned.

    def set_finish_point(self, key: str) -> None:

This is the exit point of the graph. When this node is called, the results will be the final result from the graph. It only has one argument:

Точка выхода из графа. При вызове заданной вершины, результат ее работы будет итоговым для графа. Принимает один параметр:

• key — название вершины, результат вызова который будет считаться итоговым результатом работы графа.

Вершину не нужно вызывать если на предыдущих шагах графа было создано ребро (условное или обычное) ведущее к зарезервированной вершине END.

from langgraph.graph import Graph

graph = Graph()

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

from langgraph.graph import END

This is a special node representing the end of the graph. This means that anything passed to this node will be the final output of the graph. It can be used in two places:

Зарезервированная вершина указывающая на завершение работы графа. Все данные, которые передаются вершине при вызове будут считаться результатом работы графа. Вершину можно использовать в двух случая:

• В качестве ключа end_key в add_edge.
• В качестве значения в структуре conditional_edge_mapping, передаваемой add_conditional_edges.

Представленные примеры содержат несколько методов, облегчающих работу с распространенными, готовыми графами и компонентами.

from langgraph.prebuilt import ToolExecutor

Вспомогательный класс для вызова инструментов. В качестве параметров класс принимает список инструментов.

tools = [...]
tool_executor = ToolExecutor(tools)

После инициализации класс дает доступ к исполняемому интерфейсу. Используйте класс для вызова инструментов. Передайте AgentAction для автоматического определения подходящего инструмента и входных данных.

from langgraph.prebuilt import chat_agent_executor

Вспомогательная функция для создания графа, который работает с генеративной моделью и может вызывать функции. Для использования функции передайте на вход модель и список инструментов. Модель должна поддерживать интерфейс вызова функций аналогичный OpenAI.

from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
from langgraph.prebuilt import chat_agent_executor
from langchain_core.messages import HumanMessage

tools = [TavilySearchResults(max_results=1)]
model = ChatOpenAI()

app = chat_agent_executor.create_function_calling_executor(model, tools)

inputs = {"messages": [HumanMessage(content="какая погода в саратове")]}
for s in app.stream(inputs):
    print(list(s.values())[0])
    print("----")

from langgraph.prebuilt import chat_agent_executor

This is a helper function for creating a graph that works with a chat model that utilizes tool calling. Can be created by passing in a model and a list of tools. The model must be one that supports OpenAI tool calling.

from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
from langgraph.prebuilt import chat_agent_executor
from langchain_core.messages import HumanMessage

tools = [TavilySearchResults(max_results=1)]
model = ChatOpenAI()

app = chat_agent_executor.create_tool_calling_executor(model, tools)

inputs = {"messages": [HumanMessage(content="what is the weather in sf")]}
for s in app.stream(inputs):
    print(list(s.values())[0])
    print("----")

from langgraph.prebuilt import create_agent_executor

Вспомогательная функция для работы с агентами LangChain. Для использования функции передайте на вход агента и список инструментов.

from langgraph.prebuilt import create_agent_executor
from langchain_openai import ChatOpenAI
from langchain import hub
from langchain.agents import create_openai_functions_agent
from langchain_community.tools.tavily_search import TavilySearchResults

tools = [TavilySearchResults(max_results=1)]

# Подключите шаблон промпта. Вы можете выбрать любой шаблон
prompt = hub.pull("hwchase17/openai-functions-agent")

# Выберите модель, с которой будет работать агент
llm = ChatOpenAI(model="gpt-3.5-turbo-1106")

# Создайте агента OpenAI Functions
agent_runnable = create_openai_functions_agent(llm, tools, prompt)

app = create_agent_executor(agent_runnable, tools)

inputs = {"input": "what is the weather in sf", "chat_history": []}
for s in app.stream(inputs):
    print(list(s.values())[0])
    print("----")