AI agentowe z LangChain — Część 7: Budowanie systemów wieloagentowych
Część 7 sztucznej inteligencji agentowej z LangChain: systemy wieloagentowe w LangGraph, jak agenci dzielą pracę, przekazują stan oraz są koordynowani przez węzeł nadzorczy.
Źródła: Agentic AI with LangChain — Part 7: Building Multi-Agent Systems with LangGraph
To przewodnictwo pokazuje, jak przejść od surowców do działającego systemu w przypadku: Agentic AI z LangChain — Część 7: Budowanie systemów wieloagentowych za pomocą LangGraph. Skupiamy się na krokach operacyjnych, wyraźnych sprawdzeniach oraz kodzie, który można bez problemu dodać do repozytorium, nie musząc zgadywać intencji. Na etapie przeglądu należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy udokumentować zarówno prawidłowy przebieg procesu, jak i ścieżkę naprawczą. Próby ponownych działań, kontrola przez ludzi oraz obsługa wiadomości błędnych stanowią część produktu, a nie elementy dodawane później.
Dlaczego systemy wieloagentowe są ważne
Gdy przechodzisz przez etap „Dlaczego ważni są wieloagentowi systemy”, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Wolij małe, testowalne jednostki nad rozbudowane skrypty. Gdy jakiś krok się nie powiedzie, błąd powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Ustaw punkty kontrolne po kosztownych krokach. System powinien unikać ponownego pobierania opłat za tę samą funkcję LLM, gdy operator próbuje ponownie wykonać późniejszy element.
Koncepcje techniczne użyte w tym artykule
Gdy pracujesz nad Technicznymi Koncepcjami Wykorzystywanymi na tym etapie, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzuć możliwość cichego, częściowego ukończenia zadania. Ustal punkty kontrolne po kosztownych krokach. System powrotu nie powinien ponownie naliczać opłat za tę samą wywołanie LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł.
Przegląd projektu: Agent Asystenta Finansowego
Gdy przechodzisz przez etap Projekt Overview A Financial, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zapisz czas trwania operacji oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy proces przechodzi z środowiska demonstracyjnego do współdzielonych środowisk. Ustal punkty kontrolne po drogach o wysokim koszcie. Narzędzie do kontynuacji pracy nie powinno ponownie naliczać opłat za tę samą operację LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł. Gdy przechodzisz przez etap Projekt Overview A Financial, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zdokumentuj zarówno drogę prawidłowego działania, jak i drogę naprawczą. Próby ponownych uruchomień, kontrolne punkty ludzkie oraz obsługa wiadomości błędowych są częścią produktu, a nie elementami dodatkowej optymalizacji.
1. Konfiguracja systemu
Faza konfiguracji systemu nr 1 działa najlepiej, gdy traktuje się ją jako mierzalną powierzchnię. Zapisz jeden idealny przepis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres pracy. Wolno preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakaś etapa zawiedzie, awaria powinna wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wplecione struktury ukrywają informację o tym, który węzeł zapisał dane do którego pola, i powodują przerwanie kontynuacji po przerwach.
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. Struktura projektu
Faza 2 „Struktura projektu” działa najlepiej, gdy traktuje się ją jako mierzalną powierzchnię. Zapisz jeden idealny przepis działania, jeden przypadek niepowodzenia oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres pracy. Traktuj tę fazę jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzuć ciche, częściowe ukończenie zadań. Utrzymuj stan grafu w prostej formie i określonej typowości. Wkładki nawiasowe ukrywają informację o tym, który węzeł zapisał dane w danym polu, co powoduje przerwanie kontynuacji po przerwach.
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. Narzędzia agenta asystenta finansowego
Narzędzia „3 Tools of the stage” działają najlepiej, gdy traktuje się je jako mierzalną powierzchnię do analizy. Zapisz jeden idealny zapis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres pracy. Zapisuj czasy wykonywania operacji oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy proces przechodzi z środowiska demonstracyjnego do współdzielonych środowisk. Ujawnij narzędzia o wąskich schematach oraz z wyraźnymi etykietami opisującymi efekty uboczne. Hostowie muszą wiedzieć, które wywołania zmieniają stan systemu, zanim automatycznie je zatwierdzą. Narzędzia „3 Tools of the stage” działają najlepiej, gdy traktuje się je jako mierzalną powierzchnię do analizy. Zapisz jeden idealny zapis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzysz zakres pracy. Zdokumentuj zarówno ścieżkę prawidłowego działania, jak i ścieżkę naprawczą. Próby ponownych wywołań, kontrolne punkty ludzkie oraz obsługa wiadomości błędowych stanowią część produktu, a nie elementy dodawane później.
3.1. Narzędzie do wyszukiwania danych z bazy
W etapie 3 1 Stock Data należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy preferować małe, łatwe do przetestowania jednostki nad rozbudowanymi skryptami. Gdy jakiś krok zawiedzie, powinno to wskazywać na konkretną odpowiedzialność, a nie na skomplikowaną strukturę przepływu danych. Autoryzuj się przy bramie wejściowej, a ponownie udziel uprawnień na poziomie warstwy danych. Sam token nie stanowi granicy między poszczególnymi użytkownikami.
# 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. Narzędzie do wykonywania kodu Pythona
W fazie kodu Pythona 3.2 należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Traktuj tę fazę jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nazwij pliki artefaktów, zdefiniuj sprawdzenia sukcesu i odrzuć ciche, częściowe ukończenie zadania. Oddziel budowę klienta od pętli komunikatów, aby można było wymieniać dostawców bez konieczności przepisywania maszyny stanu rozmowy.
@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. Narzędzie do wyszukiwania na Wikipedii
Dla etapu wyszukiwania w Wikipedii 3 3 należy zdefiniować dane wejściowe, osobę odpowiedzialną za ten krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy rejestrować czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy proces przechodzi z środowiska demonstracyjnego do współdzielonych środowisk. Należy uwierzytelniać się przy bramie wejściowej oraz ponownie autoryzować w warstwie danych. Sam token nie stanowi granicy między poszczególnymi użytkownikami. Dla etapu wyszukiwania w Wikipedii 3 3 należy zdefiniować dane wejściowe, osobę odpowiedzialną za ten krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić ten krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy udokumentować zarówno prawidłowy przebieg procesu, jak i ścieżki naprawcze. Próby ponownych działań, mechanizmy kontroli ludzkiej oraz obsługa wiadomości błędowych stanowią część produktu, a nie elementy dodawane później.
@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. Swarm Multi-Agents
Podczas przechodzenia przez etap 4 Swarm Multi-Agents najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Wolno preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok się nie powiedzie, błąd powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowaną sekwencję działań. Ustaw punkty kontrolne po kosztownych krokach. Narzędzie do kontynuacji pracy nie powinno ponownie naliczać opłat za tę samą wywołanie LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł.
4.1. Architektura Swarm Multi-Agents
Gdy pracujesz nad architekturą 4 1 etapu, najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzuć ciche, częściowe ukończenie zadania. Ustaw punkty kontrolne po kosztownych krokach. Program powinien unikać ponownego pobierania opłat za tę samą funkcję LLM, gdy operator próbuje ponownie wykonać późniejszy węzeł.
4.2. Implementacja wielu agentów w swarmie
Podczas pracy nad etapem 4 2 Swarm Multi-agents najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zapisz czasy wykonywania zadań oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy przechodzi się od środowiska demonstracyjnego do wspólnych środowisk. Ustal punkty kontrolne po kosztownych krokach. Narzędzie do kontynuacji pracy nie powinno ponownie naliczać opłat za tę samą wywołanie LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł. Podczas pracy nad etapem 4 2 Swarm Multi-agents najpierw zapisz umowę: wymagane dane wejściowe, sygnał sukcesu oraz to, co dzieje się w przypadku częściowego niepowodzenia. Taka lista kontrolna zapewnia uczciwość późniejszych zmian w kodzie. Zdokumentuj zarówno ścieżkę prawidłowego działania, jak i ścieżkę naprawczą. Próby ponownego uruchomienia, kontrolne punkty ludzkie oraz obsługa wiadomości błędnych stanowią część produktu, a nie elementy dodatkowej optymalizacji.
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
Faza 5 Supervisor Multi-Agents funkcjonuje najlepiej, gdy traktowana jest jako mierzalna struktura. Zapisz jeden idealny przykład działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres projektu. Wolno preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok zawiedzie, awaria powinna wskazywać na konkretną odpowiedzialność, a nie na skomplikowaną sekwencję działań. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wtórne struktury ukrywają informację o tym, który węzeł zapisał dane do którego pola, i powodują przerwanie kontynuacji działania po zakłóceniach.
5.1. Architektura Supervisor Multi-Agents
Architektura 5 1 dla etapów pracy funkcjonuje najlepiej, gdy traktowana jest jako mierzalna powierzchnia. Zapisz jeden „złoty” przepis, jeden przypadek niepowodzenia oraz notatkę o cofnięciu działań, zanim rozszerzysz zakres pracy. Traktuj ten etap jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzucaj ciche, częściowe ukończenie zadań. Utrzymuj stan grafu w formie prostych, spójnych struktur typowych. Wkładki nawiasowe ukrywają informację o tym, który węzeł zapisał dane do którego pola, co powoduje przerwanie kontynuacji pracy po przerwach.
5.2. Implementacja wielu agentów nadzorczych
Faza 5 2 Supervisor Multi-agents funkcjonuje najlepiej, gdy jest traktowana jako mierzalna powierzchnia. Zapisz jeden idealny zapis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres. Zapisuj czasy wykonywania operacji oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy ścieżka przechodzi z wersji demonstracyjnej do środowisk współdzielonych. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wplecione elementy ukrywają informację o tym, który węzeł zapisał dane w danym polu, i powodują przerwanie kontynuacji działania po przerwach. Faza 5 2 Supervisor Multi-agents funkcjonuje najlepiej, gdy jest traktowana jako mierzalna powierzchnia. Zapisz jeden idealny zapis działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia działań, zanim rozszerzysz zakres. Zdokumentuj zarówno ścieżkę prawidłowego działania, jak i ścieżkę przywracania do normalnego stanu. Próby ponownych działań, kontrolne punkty ludzkie oraz obsługa wiadomości nieodebranych stanowią część produktu, a nie elementy dodawane później.
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. Podsumowanie końcowe
W fazie 6 „Podsumowanie końcowe” należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy dany krok zawiedzie, powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany proces. Konieczne jest ludzkie zatwierdzenie w przypadkach, gdy dochodzi do wydatków lub zmian w danych produkcyjnych. Połączenia realizowane w czasie kompilacji nie równają się kompletności biznesowej.
Lista kontrolna operacyjna
Faza „Lista kontrolna operacyjna” działa najlepiej, gdy jest traktowana jako mierzalna powierzchnia do analizy. Należy zebrać jeden idealny przykład działania, jeden przypadek awarii oraz notatkę dotyczącą cofnięcia zmian, zanim rozszerzy się zakres pracy.
Zachowaj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności przeglądania całej struktury.
Zachowaj stan grafu w formie prostych, typizowanych elementów. Wplecione struktury ukrywają informację o tym, który węzeł zapisał dane do którego pola, i utrudniają kontynuację pracy po przerwach.
Gdy budżet na to pozwala, dodaj test sprawdzający kluczową ścieżkę w procesie CI przy użyciu narzędzi testowych, a nie rzeczywistych, płatnych API.
Dokumentuj zarówno prawidłową ścieżkę działania, jak i ścieżkę przywracania do normalnego funkcjonowania. Próby ponownych działań, kontrola przez ludzi oraz obsługa wiadomości błędowych stanowią część produktu, a nie elementy dodawane później.
Zachowaj stan grafu w formie prostych, typizowanych elementów. Wplecione struktury ukrywają informację o tym, który węzeł zapisał dane do którego pola, i utrudniają kontynuację pracy po przerwach.
Literatura pokrewna
- Praktyczne notatki: Budowanie systemu triagowania zamówień podtrzymywania relacji między wieloma agentami za pomocą LangGraph — Szczegółowy przewodnik po Praktycznych notatkach: Budowanie systemu triagowania zamówień podtrzymywania relacji między wieloma agentami za pomocą LangGraph: kontrakty, sprawdzenia oraz miejsca na kod do wklejenia dla zespołów wdrażających ten wzorzec.
- Praktyczne notatki: Systemy wielu agentów LangGraph: Od jednego mózgu do wielu — Szczegółowy przewodnik po Praktycznych notatkach: Systemy wielu agentów LangGraph: Od jednego mózgu do wielu: kontrakty, sprawdzenia oraz miejsca na kod do wklejenia dla zespołów wdrażających ten wzorzec.