Агентные ИИ с LangChain — Часть 7: Создание многопроцессорных агентов

В этом руководстве пошагово описывается процесс создания системы от сырья до готового решения для: Агентные ИИ с LangChain — Часть 7: Создание многоагентных систем с LangGraph. Основное внимание уделяется практическим шагам, четкой проверке результатов и коду, который можно просто добавить в репозиторий без необходимости догадываться о его назначении. На этапе обзора необходимо определить входные данные, ответственного за выполнение шага и критерии завершения перед внесением изменений в код. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии системы. Необходимо одновременно задокументировать успешный сценарий выполнения и сценарий восстановления. Повторные попытки, проверка человеком и обработка ошибок являются неотъемлемой частью продукта, а не элементами последующей доработки.

Почему важны многоагентные системы

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

Технические концепции, используемые в этой статье

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

Обзор проекта: Агент-финансовый ассистент

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

1. Конфигурация системы

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

OPENAI_API_KEY="<Your OpenAI API Key>"
LANGSMITH_TRACING=true
LANGSMITH_ENDPOINT="https://api.smith.langchain.com"
LANGSMITH_API_KEY="<Your LangSmith API Key"
LANGSMITH_PROJECT="agent-ai"

class BaseConfig(BaseSettings):
    OPENAI_API_KEY: Optional[str]

model_config = SettingsConfigDict(env_file=".env", extra="ignore")

2. Структура проекта

Этап структурирования проекта «2» работает наилучшим образом, если рассматривать его как измеримую основу. Соберите один идеальный пример результата, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объема работ. Рассматривайте этот этап как контракт между входными данными и проверенными результатами. Дайте названия элементам, определите критерии успеха и не соглашайтесь на молчаливое частичное выполнение задач. Сохраняйте структуру графа простой и типизированной. Вложенные структуры скрывают информацию о том, какой узел заполнил тот или иной поле, и мешают возобновлению работы после прерываний.

project/
├── tools/
│   ├── __init__.py
│   ├── financial_search.py # fetch stock prices for companies
│   ├── python_REPL.py      # async wrapper of PythonREPL()
|   └── wikipedia_search.py # fetch financial summary of companies
│
├── graphs/            # implement subagents and multi-agents
│   ├── agent.py       # implement research and analyst in a single agent
│   ├── sub_agents.py  # implement research and analyst sub agents
│   ├── multi_agent_supervisor.py # implement supervisor multi-agents
│   └── multi_agent_swarm.py # implement swamp multi-agents
│
├── config.py           # pydantic BaseConfig
│── utils.py            # utility for printing and executing agents
├── .env                # environment variable definition
├── .gitignore
└── requirements.txt     # package requirements

3. Инструменты агента финансового помощника

Инструменты «3 Tools of the stage» работают наилучшим образом, когда их рассматривают как измеримую среду. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объёма работ. Записывайте временные показатели, а также стоимость токенов или запросов рядом с функциональными результатами. Отображение стоимости на раннем этапе предотвращает неожиданные счёты при переходе от демо-среды к общедоступным средам. Используйте инструменты с узкими схемами и чёткими метками о побочных эффектах. Администраторам необходимо знать, какие вызовы изменяют состояние системы, прежде чем они автоматически одобрят их. Инструменты «3 Tools of the stage» работают наилучшим образом, когда их рассматривают как измеримую среду. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объёма работ. Документируйте одновременно успешный и восстановительный пути работы. Повторные попытки, проверки человеком и обработка неработающих сообщений являются частью продукта, а не элементами последующей доработки.

3.1. Инструмент поиска статических данных

Для этапа обработки данных 3 1 Stock Data необходимо заранее определить входные данные, ответственного за выполнение шага и критерии завершения перед внесением изменений в код. Операторы должны иметь возможность перезапустить шаг, исходя из известной точки контроля, без необходимости угадывать скрытое состояние. Лучше использовать небольшие, тестируемые единицы кода вместо обширных скриптов. При сбое шага причина должна быть связана с конкретной функцией, а не с запутанной структурой обработки данных. Аутентификация должна происходить на уровне шлюза, а повторная авторизация — на уровне обработки данных. Одного лишь токена не достаточно для определения границы тенантности.

# define LangChain tool to retrieve stock data
@tool
async def stock_data_tool(
        company_ticker: Annotated[str, "The ticker symbol of the company to retrieve their stock performance data."],
        num_days: Annotated[int, "The number of business days of stock data required to respond to the user query."]
) -> str:

        # fetch stock data asynchronously from yfinance
        # check full code for _fetch_stock_sync function
        df = await asyncio.to_thread(_fetch_stock_sync, company_ticker, f"{num_days}d")

        return (f"Successfully executed the stock performance data retrieval tool to retrieve "
                f"the last *{num_days} days* of data for company **{company_ticker}**:\n\n{df.to_markdown()}")

3.2. Инструмент выполнения кода на Python

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

@tool
async def python_repl_tool(
        code: Annotated[str, "The python code to execute to generate your chart."],
) -> str:

    # wrap LLM generated code and execute it by python.REPL()
    # repl = PythonREPL().
    clean_code = textwrap.dedent(code).strip()
    result = await asyncio.to_thread(repl.run, clean_code)

    return (
        f"Successfully executed the Python REPL tool.\n\n"
        f"Python code executed:\n```python\n{code}\n```\n\n"
        f"Code output:\n```\n{result}```"
    )

3.3. Инструмент поиска в Википедии

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

@tool()
async def wikipedia_tool(
        query: Annotated[str, "The Wikipedia search to execute to find key summary information."],
) -> str:
    """Use this to search Wikipedia for factual information asynchronously."""
    headers = {"User-Agent": "MyLangChainAgent/1.0 (contact@example.com)"}

    async with httpx.AsyncClient(headers=headers, follow_redirects=True) as client:

        # Step 1: Search Wikipedia for matching pages
        search_url = "https://en.wikipedia.org/w/api.php"
        search_params = {
            "action": "query",
            "list": "search",
            "srsearch": query,
            "format": "json"
        }
        response = await client.get(search_url, params=search_params)
        search_data = response.json()
        results = search_data.get("query", {}).get("search", [])
        title = results[0]["title"]

        # Step 2: Fetch the page extract/summary
        summary_url = f"https://en.wikipedia.org/api/rest_v1/page/summary/{title}"
        summary_res = await client.get(summary_url)
        summary_data = summary_res.json()
        summary = summary_data.get("extract", "No summary available.")

        return f"Successfully executed:\nWikipedia summary: {summary}"

4. Множественные агенты стаи

При работе над этапом 4 «Множественные агенты стаи» сначала запишите контракт: необходимые входные данные, сигнал о успехе и действия при частичной неудаче. Такой список поможет сохранять честность при последующих изменениях кода. Предпочитайте небольшие, тестируемые единицы кода вместо обширных скриптов. Если какой-то шаг не сработает, причина неудачи должна указывать на конкретную ответственность, а не на запутанную цепочку операций. Создавайте контрольные точки после дорогостоящих шагов. Функция возобновления работы не должна повторно запускать один и тот же вызов большой языковой модели, когда оператор пытается выполнить следующий узел.

4.1. Архитектура множественных агентов стаи

При работе над архитектурой этапа 4.1 сначала запишите условия контракта: необходимые входные данные, сигнал о успешном выполнении и действия при частичной неудаче. Такой список помогает сохранять честность при последующих изменениях кода. Рассматривайте этот этап как контракт между входными данными и проверенными выходными результатами. Дайте названия элементам, определите критерии успешности и не допускайте молчаливого частичного выполнения задачи. Выполняйте контрольные точки после дорогостоящих операций. Система возобновления работы не должна повторно взимать плату за один и тот же вызов большой языковой модели, когда оператор пытается выполнить следующий элемент.

4.2. Реализация многокомпонентной системы Swarm

При работе над этапом 4 2 Swarm Multi-agents сначала запишите условия взаимодействия: необходимые входные данные, сигнал о успехе и действия при частичной неудаче. Такой чек-лист поможет сохранять честность при последующих изменениях кода. Записывайте время выполнения и стоимость токенов или запросов рядом с функциональными результатами. Отображение затрат заранее предотвращает неожиданные счета при переходе от демо-среды к общедоступным средам. Создавайте контрольные точки после дорогостоящих шагов. Функция возобновления работы не должна снова взимать плату за один и тот же вызов LLM при повторной попытке обработки последующего узла. При работе над этапом 4 2 Swarm Multi-agents сначала запишите условия взаимодействия: необходимые входные данные, сигнал о успехе и действия при частичной неудаче. Такой чек-лист поможет сохранять честность при последующих изменениях кода. Документируйте как успешный, так и путь восстановления. Повторные попытки, проверки человеком и обработка неработающих сообщений являются частью продукта, а не элементами последующей доработки.

llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini")

# Create a handoff tool for researcher -> analyst
transfer_to_analyst = create_handoff_tool(
    agent_name="analyst",
    description="Transfer user to the analyst assistant, who can create visualizations of provided data.",
)

# create research agent as a subagent
research_agent = create_agent(
    llm,
    tools=[wikipedia_tool, stock_data_tool, transfer_to_analyst],
    system_prompt="You provide summaries from Wikipedia, and can query load raw, numerical stock performance data.",
    name="researcher"
)

# define the checkpointer
checkpointer = InMemorySaver()

# create swarm agents
swarm_agent = create_swarm(
    agents=[research_agent, analyst_agent],
    default_active_agent="researcher"
).compile(checkpointer=checkpointer)

5. Supervisor Multi-Agents

Этап 5 «Supervisor Multi-Agents» работает наилучшим образом, если рассматривать его как измеримую структуру. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объёма задачи. Предпочитайте небольшие, тестируемые единицы кода вместо обширных скриптов. При сбое какого-либо шага причина должна быть связана с конкретной ответственностью, а не с запутанной цепочкой операций. Сохраняйте структуру графа простой и типизированной. Вложенные структуры данных маскируют информацию о том, какой узел заполнил тот или иной поле, что приводит к нарушению работоспособности после перерывов.

5.1. Архитектура Supervisor Multi-Agents

Архитектура 5 1 для этапных работ наилучшим образом функционирует, когда её рассматривают как измеримую поверхность. Запишите один идеальный пример выполнения, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объёма работ. Рассматривайте этот этап как контракт между входными данными и проверенными выходными результатами. Дайте названия создаваемым объектам, определите критерии успешного выполнения и не соглашайтесь на молчаливое частичное завершение работ. Сохраняйте структуру графа простой и типизированной. Вложенные структуры данных скрывают информацию о том, какой узел заполнил тот или иной поле, что приводит к нарушению возобновления работы после перерывов.

5.2. Реализация с системой множества агентов-супервайзеров

Этап «5 2 Supervisor Multi-agents» работает наилучшим образом, когда его рассматривают как измеримую поверхность. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объёма работ. Записывайте временные показатели, а также стоимость токенов или запросов рядом с функциональными результатами. Отображение стоимости на раннем этапе предотвращает неожиданные счёты при переходе от демо-среды к общедоступным средам. Сохраняйте структуру графа простой и типизированной. Вложенные структуры скрывают информацию о том, какой узел заполнил тот или иной поле, и нарушают возможность продолжения работы после прерываний. Этап «5 2 Supervisor Multi-agents» работает наилучшим образом, когда его рассматривают как измеримую поверхность. Соберите один идеальный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объёма работ. Документируйте одновременно успешный путь выполнения и путь восстановления. Повторные попытки, проверки человеком и обработка неработающих сообщений являются частью продукта, а не элементами последующей доработки.

llm = ChatOpenAI(api_key=api_key, model="gpt-4o-mini")

# create a checkpointer
checkpointer = InMemorySaver()

# create research subagent
research_agent = create_agent(
    llm,
    tools=[wikipedia_tool, stock_data_tool],
    system_prompt="You provide summaries from Wikipedia, and can query load raw, numerical stock performance data.",
    name="researcher"
)

# Create the supervisor multi-agent graph and compile it
supervisor = create_supervisor(
    model=llm,
    agents=[research_agent, analyst_agent],
    prompt=(
        "You are a supervisor managing two agents:\n"
        "- a research agent. Assign research and data collection tasks to this agent\n"
        "- an analyst agent. Assign the creation of visualizations via code to this agent\n"
        "Assign work to one agent at a time, do not call agents in parallel.\n"
        "Do not do any work yourself."
    ),
    add_handoff_back_messages=True,
    output_mode="last_message"
).compile(checkpointer=checkpointer)

6. Заключительное резюме

На этапе 6 «Заключительный обзор» необходимо определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии системы. Лучше использовать небольшие, тестируемые единицы кода вместо обширных скриптов. При сбое шага он должен указывать на конкретную причину, а не на сложную структуру обработки данных. Внедрять человеческое утверждение для операций, связанных с расходованием средств или изменением производственных данных. Наличие связей во время компиляции не гарантирует полноты бизнес-логики.

Чек-лист операций

Этап чек-листа операций работает наилучшим образом, когда рассматривается как измеримая структура. Соберите один эталонный пример работы, один случай сбоя и записку о возврате к предыдущему состоянию перед расширением объема работ.

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

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

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

Документируйте как успешный, так и восстановительный пути работы. Повторные попытки, проверки человеком и обработка неработоспособных сообщений являются частью продукта, а не последующими улучшениями.

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