Агентський ШІ з LangChain — Частина 7: Створення багатоагентних систем
Частина 7 про агентські ШІ з LangChain: багатоагентні системи в LangGraph, як агенти ділять роботу, передають стан та координуються за допомогою вузла-керівника.
Джерела: Agentic AI with LangChain — Part 7: Building Multi-Agent Systems with LangGraph
У цьому посібнику детально описано шлях від сировини до функціональної системи для: Агентного ШІ з LangChain — Частина 7: Створення багатоагентних систем з LangGraph. Основна увага приділяється конкретним крокам виконання, чітким перевіркам та коду, який можна просто додати до репозиторію, не намагаючись здогадатися про мету. На етапі огляду необхідно визначити вхідні дані, виконавця кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись зрозуміти прихований стан. Необхідно документувати як успішний, так і відновлювальний сценарії роботи. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації.
Чому важливі багатоагентні системи
Під час роботи над етапом «Чому важливі багатоагентні системи» спочатку запишіть контракт: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Такий перелік допомагає зберігати чесність пізніших змін у коді. Віддавайте перевагу невеликим, тестованим одиницям перед об’ємними скриптами. Коли якийсь крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на заплутану послідовність операцій. Робіть контрольні пункти після дорогих кроків. Система відновлення не повинна знову стягувати плату за один і той самий виклик LLM, коли оператор намагається виконати пізнішу операцію.
Технічні концепції, використані у цій статті
Під час роботи над технічними концепціями, які використовуються на цьому етапі, спочатку запишіть умови контракту: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Такий перелік допомагає зберігати чесність пізніших змін у коді. Розглядайте цей етап як контракт між вхідними даними та перевіреними результатами. Позначте всі елементи, визначте критерії успіху та не допускайте безпроблемного часткового виконання завдань. Робіть перевірки після дорогих кроків. Система повернення до виконання не повинна знову стягувати плату за один і той самий виклик LLM, коли оператор намагається виконати наступний етап.
Огляд проекту: Агент фінансового асистента
Під час роботи над етапом «Огляд проекту: фінанси» спочатку запишіть умови використання: необхідні дані вхіду, сигнал про успіх та наслідки часткової невдачі. Цей перелік допомагає зберігати чесність пізніших змін у коді. Запишіть час виконання та витрати на токени або запити поруч із результатами функціоналу. Чітке бачення витрат заздалегідь запобігає несподіваним рахункам під час переходу від демо-версії до спільних середовищ. Робіть контрольні пункти після дорогих кроків. Система не повинна знову стягувати плату за один і той самий виклик ШІ, якщо оператор перезапускає пізніший етап. Під час роботи над етапом «Огляд проекту: фінанси» спочатку запишіть умови використання: необхідні дані вхіду, сигнал про успіх та наслідки часткової невдачі. Цей перелік допомагає зберігати чесність пізніших змін у коді. Одночасно задокументуйте оптимальний та резервний сценарії роботи. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації.
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 необхідно визначити вхідні дані, виконавця кроку та критерії завершення перед зміною коду. Оператори повинні мати можливість перезапустити крок з відомої точки контролю, не намагаючись вгадати прихований стан. Розглядайте цей етап як контракт між вхідними даними та перевіреними результатами. Призначте назви artefaktам, визначте критерії успіху та не допускайте беззвучного часткового завершення. Відокремте процес створення клієнта від циклу обробки повідомлень, щоб можна було замінювати постачальників без переписування машини станів розмови.
@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-м етапом «Багатоагентні системи зграї» спочатку запишіть контракт: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Такий перелік допомагає зберігати чесність пізніших змін у коді. Віддавайте перевагу невеликим, тестованим одиницям перед об’ємними скриптами. Коли якийсь крок зазнає невдачі, причина має вказувати на конкретну відповідальність, а не на заплутану послідовність операцій. Робіть контрольні пункти після дорогих кроків. Функція відновлення не повинна знову оплачувати один і той самий виклик LLM, коли оператор намагається виконати пізніший етап.
4.1. Архітектура багатоагентних систем зграї
Під час роботи над архітектурою 4 1 етапу спочатку запишіть умови контракту: необхідні вхідні дані, сигнал про успіх та те, що відбувається у разі часткової невдачі. Цей перелік допомагає зберігати чесність пізніших змін у коді. Розглядайте цей етап як контракт між вхідними даними та перевіреними результатами. Позначте всі елементи, визначте критерії успіху та не допускайте безповідомного часткового виконання. Робіть перевірки після дорогих кроків. Система відновлення не повинна знову оплачувати один і той самий виклик LLM, коли оператор намагається виконати наступний етап.
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.
Документуйте як шлях успішної роботи, так і шлях відновлення одночасно. Повторні спроби, людський контроль та обробка некоректних повідомлень є частиною продукту, а не етапом подальшої оптимізації.
Зберігайте стан графа у вигляді плоскої структури з типизацією. Вкладені блоки приховують інформацію про те, який вузол заповнив яке поле, і ускладнюють продовження роботи після перерв.