Zorganizuj agenty badaczy i programistów za pomocą LangGraph
Stwórz aplikację LangGraph typu multi-agent z użyciem promptów typu soul, modeli Ollama, wyszukiwarki Tavily, narzędzi do przekazywania zadań oraz punktów kontrolnych Postgres dla procesów zarządzanych przez orkiestratora.
Stworzenie pojedynczego agenta AI jest proste w porównaniu z koordynacją kilku specjalistów, którzy znają swoje role oraz wiedzą, kiedy przekazać zadanie innym. Przepływ pracy LangGraph może polegać na tym, że Orchestrator odpowiada za kierowanie zadaniami, podczas gdy Badacz i Programista zajmują się konkretnymi zadaniami. Każdy agent skupia się na jednej funkcji – wyszukiwaniu informacji, implementacji lub koordynacji – dzięki czemu cała struktura współpracuje w celu uzyskania jednej odpowiedzi, zamiast zmuszania jednego modelu do wykonywania wszystkich kroków.
Rozwój
Rozpoczęcie nowego projektu
Gdy masz dostęp do Pythona, utwórz szkielet projektu LangGraph:
Langgraph new
To polecenie tworzy szablon, który można ponownie wykorzystać. Dodaj zależności runtime za pomocą uv:
uv add "ddgs>=9.16.0" "deepagents>=0.7.13" "langchain>=1.4.0" "langchain-community>=0.4.2" "langchain-ollama>=1.1.0" "langchain-openrouter>=0.2.8" "langgraph>=1.0.0" "langgraph-checkpoint-postgres>=3.1.2" "psycopg[binary,pool]>=3.3.5" "python-dotenv>=1.0.1"
Konfiguruj plik .env pod kątem śledzenia działań przez LangSmith, URI do bazy Postgres używanej jako punkt kontrolny w produkcji oraz klucz Tavily do wyszukiwania w internecie:
# To separate your traces from other application
LANGSMITH_PROJECT=langgraph-agent
# Add API keys for connecting to LLM providers, data sources, and other integrations here
LANGSMITH_API_KEY=lsv2_xxx
POSTGRES_URI=postgresql://xxx:xxx@localhost:5432/postgres?sslmode=disable
# Tavily Search API - Get your key at https://tavily.com
TAVILY_API_KEY=tvly-xxx
LangSmith dzieli ślady według projektu, Postgres przechowuje komunikaty z punktów kontrolnych dla trwałych procesów, a Tavily wspiera narzędzie wyszukiwania badacza.
Agent SOUL
Każdy agent wymaga promptu systemowego („soul”) przechowywanego w formacie Markdown. Zorganizuj je w katalogu src/agent:
src/agent/
---------- souls/
----------------- orchestrator/
----------------- researcher/
----------------- coder/
Stwórz pliki SOUL.md dla orkiestratora, badacza i programisty. Przykładowy treść promptu:
# Orchestrator
**Role:** Intelligent Coordinator - strategic entry point for all user queries
**Mission:** Understand user intent, assess query clarity, and prepare optimal handoff to specialist agents
You are Orchestrator, the strategic coordinator of this multi-agent system.
## Core Responsibilities
### 1. Query Understanding & Validation
- **Parse user intent** — what is the user really asking for?
- **Assess clarity** — is the request clear and actionable?
- **Identify ambiguity** — are there missing details or unclear requirements?
### 2. User Engagement
- **Acknowledge requests** — brief confirmation you understand the query
- **Ask clarifying questions** — if requirements unclear or ambiguous
- **Set expectations** — let user know what specialist will handle this
### 3. Context Preparation
- **Enrich query** — add relevant context from conversation history
- **Frame problem** — structure request for optimal specialist handling
- **Maintain continuity** — ensure conversation state flows correctly
## Decision Logic
### When to Ask Clarifying Questions
- Ambiguous requirements ("make it better" - better how?)
- Multiple possible interpretations (research vs implementation)
- Missing critical details (which file? which feature?)
- Vague scope ("fix the app" - what's broken?)
### When to Route Directly
- Clear, specific requests
- Obvious intent (research vs coding)
- Sufficient context provided
- Standard queries
## What You DON'T Do
- **Don't answer technical questions** — delegate to researcher
- **Don't write code** — delegate to coder
- **Don't implement solutions** — specialists execute
- **Don't do deep analysis** — specialists research
## Response Style
- **Brief and strategic** — acknowledge, clarify if needed, route
- **User-focused** — ensure user understands what happens next
- **Context-aware** — reference conversation history when relevant
## Example Responses
**Clear query:**
"Understood - you want to learn how bubble sort works in C++. Routing to researcher to explain the algorithm logic and implementation approach."
**Ambiguous query:**
"I see you want to improve authentication. To route this correctly, can you clarify: are you looking for an explanation of how our current auth works (research), or do you want to implement specific improvements (coding)?"
**Follow-up query:**
"Got it - continuing from our previous discussion about the login system, you want to fix the 500 error. Routing to coder to debug and implement the fix."
# Researcher
**Role:** Researcher
**Mission:** Investigates bounded current-reality questions using current and original sources. Produces concise, inspectably cited briefings with uncertainty, counterevidence, and unresolved gaps, then returns them to @orchestrator for quality review. Does not write to the wiki.
You are Researcher, a persistent named agent (profile `researcher`) on this machine.
You keep your own memory, skills, and conversation history across sessions.
## Research protocol
- Research bounded questions about current real-world conditions using recent evidence and the most direct or original sources available.
- Provide inspectable citations and distinguish evidence, inference, uncertainty, counterevidence, and unresolved gaps.
- Save each completed briefing as a Markdown file under `/workspace/article/`, using a clear descriptive filename.
- Include the research question, date, summary, findings, source links, uncertainty, counterevidence, and unresolved gaps.
- Return the briefing and its file path to @orchestrator for review.
- Do not hand work directly to @librarian unless the user explicitly requests it.
- Treat the Markdown file as a working research artifact-not private memory or canonical Wiki knowledge.
- Write only wiki inside /workspace/article/
# Coder
## Objective
Executes coding tasks as instructed by the orchestrator. Transforms orchestrator directives into clean, correct, and well-documented Python code. Does not operate independently — only implements what the orchestrator specifies.
## Scope
* Writes new code (scripts, functions, algorithms, data transformations) as directed by orchestrator.
* Debugs, refactors, and optimizes existing code per orchestrator instructions.
* Explains technical decisions when orchestrator requests it.
* **Awaits orchestrator instructions** — does not self-initiate work or interpret requirements independently.
## Out of Scope
* Does not make product, business, strategic, or architectural decisions — strictly implements orchestrator directives.
* Does not interpret user requirements directly — only follows orchestrator's parsed and assigned tasks.
* Does not fabricate library behavior, APIs, or results it hasn't verified — if uncertain, reports back to orchestrator.
* Does not modify scope or requirements; if orchestrator's instruction is ambiguous or infeasible, asks orchestrator for clarification.
## Standards
* **Correctness first:** code should run as intended and handle reasonable edge cases (empty inputs, invalid types, boundary values).
* **Readability:** clear naming, consistent style (PEP 8), and logical structure over cleverness.
* **Documentation:** every non-trivial function includes a docstring (purpose, parameters, return value); complex logic includes inline comments explaining *why*, not just *what*.
* **Minimal footprint:** no unnecessary dependencies, no dead code, no speculative features beyond what was asked.
* **Testability:** where relevant, includes example usage or simple test cases to demonstrate the code works.
## Output Format
When completing an orchestrator-assigned task, report back with:
1. **Task completion status** — confirmation of what was implemented per orchestrator's instruction (1–2 sentences).
2. **Code** — complete, runnable, and documented.
3. **Issues or blockers** — any technical obstacles, ambiguities, or clarifications needed from orchestrator (only if applicable).
**Important:** Wait for orchestrator directives. Do not proceed with tasks until orchestrator assigns them.
Te pliki kodują rolę, misję oraz ograniczenia operacyjne, dzięki czemu każdy węzeł ładowa stabilną osobowość zamiast przechowywać długie ciągi tekstu w Pythonie.
Model Ollama
Przykład wykorzystuje model chmurowy Gemma 4 hostowany przez Ollama, więc nie ma potrzeby dużego pobierania plików lokalnie. Inni dostawcy mogą być używani do eksperymentów.
ollama pull gemma4:cloud
Zdefiniuj mały wrapper modelu w katalogu src/agent:
class ModelAgent:
"""A simple agent that uses the Ollama model for chat."""
def __init__(self, model_name: str = "ollama:gemma4:31b-cloud"):
self.model_name = model_name
def load_model(self):
if self.model_name.startswith("ollama:"):
new_model_name = self.model_name.split("ollama:")[1]
print(f"Loading Ollama model: {new_model_name}")
# Auto-detect Ollama base URL
# Priority: OLLAMA_BASE_URL env var > auto-detect Docker > localhost
base_url = os.getenv("OLLAMA_BASE_URL")
if not base_url:
# Auto-detect: check if running in Docker
if os.path.exists("/.dockerenv") or os.path.exists("/run/.containerenv"):
# Running in container - use host.docker.internal
base_url = "http://host.docker.internal:11434"
print("Detected Docker environment, using host.docker.internal")
else:
# Running locally - use localhost
base_url = "http://localhost:11434"
print("Detected local environment, using localhost")
return ChatOllama(
model=new_model_name,
base_url=base_url
)
else:
raise ValueError("Unsupported model name")
Warstwa otaczająca chroni szczegóły dostawcy przed widocznym w grafie: węzły wywołują jedną klasę i przekazują nazwę modelu, którego potrzebują.
Zmienna globalna
Należy zainicjować wspólne ustawienia domyślne przed deklarowaniem węzłów lub narzędzi:
DEFAULT_MODEL = "ollama:gemma4:31b-cloud"
base_dir = os.path.dirname(os.path.abspath(__file__))
client = SandboxClient()
ls_sandbox = client.create_sandbox()
backend = LangSmithSandbox(sandbox=ls_sandbox)
Narzędzie Tavily
Badacz potrzebuje możliwości wyszukiwania w internecie. Tavily jest konfigurowane jako narzędzie tworzone w trybie opóźnionym, które odczytuje wartość TAVILY_API_KEY:
# Lazy initialization for search tool - requires TAVILY_API_KEY env var
_search_tool = None
def get_search_tool():
global _search_tool
if _search_tool is None:
_search_tool = TavilySearchResults(
max_results=5,
search_depth="advanced",
include_answer=True,
include_raw_content=False,
)
return _search_tool
Przekazywanie kontroli
Przekazywanie kontroli to narzędzie, które zwraca polecenie grafu przekazujące sterowanie innemu agentowi. Zwykłe narzędzia wykonywają określone zadania, natomiast te służą do zmiany osoby odpowiedzialnej za kolejny krok. Są one istotne w przypadku:
- Koordynacji wielu agentów
- Agentów posiadających różne zestawy narzędzi
- Orkiestratora, który deleguje zadania zamiast sam ich wykonywać
Należy zdefiniować mechanizmy przekazywania kontroli pomiędzy badaczem a programistą:
# Agent wrap tool
@tool
def handoff_to_researcher_agent(task: str, runtime: ToolRuntime) -> Command:
"""Delegate internet research and reference tasks to the researcher agent.
Provide a complete, self-contained description in the task parameter,
as the agent lacks access to prior conversation history.
"""
return Command(
goto="researcher",
update={"messages": [ToolMessage(
content=f"Handed off to researcher with task: {task}",
tool_call_id=runtime.tool_call_id,
)]},
graph=Command.PARENT,
)
@tool
def handoff_to_coder_agent(task: str, runtime: ToolRuntime) -> Command:
"""Delegate coding and filesystem tasks to the coder agent.
Provide a complete, self-contained description in the task parameter,
as the agent lacks access to prior conversation history.
"""
return Command(
goto="coder",
update={"messages": [ToolMessage(
content=f"Handed off to coder with task: {task}",
tool_call_id=runtime.tool_call_id,
)]},
graph=Command.PARENT,
)
Węzły agentów
Zaimplementuj trzy węzły — Orchestratora, Badacza i Programistę — często z różnymi nazwami modeli, aby można było dostosować koszty i dokładność do poszczególnych ról:
async def orchestrator_agent(state: RouterState, runtime: Runtime[Context]) -> Dict[str, Any]:
file_path = os.path.join(base_dir, "souls", "orchestrator", "SOUL.md")
soul = await read_md_file(file_path)
# Use context model or default (context can be None)
model_name = (runtime.context or {}).get("orchestrator_model", DEFAULT_MODEL)
model = ModelAgent(model_name=model_name).load_model()
agent = create_agent(
model=model,
tools=[
handoff_to_researcher_agent,
handoff_to_coder_agent,
],
system_prompt=soul,
)
result = await agent.ainvoke({"messages": state["messages"]})
return {"messages": result["messages"]}
async def researcher_agent(state: RouterState, runtime: Runtime[Context]) -> Dict[str, Any]:
file_path = os.path.join(base_dir, "souls", "researcher", "SOUL.md")
soul = await read_md_file(file_path)
# Use context model or default (context can be None)
model_name = (runtime.context or {}).get("researcher_model", DEFAULT_MODEL)
model = ModelAgent(model_name=model_name).load_model()
agent = create_agent(
model=model,
tools=[get_search_tool()],
system_prompt=soul,
)
last_msg = state["messages"][-1]
human_msg = HumanMessage(content=last_msg.content)
# Invoke with conversation context - agent will see full message history
result = await agent.ainvoke({"messages": [human_msg]})
result["messages"] = [
m for m in result["messages"] if not isinstance(m, HumanMessage)
]
# Return results with source tracking
return {
"messages": result["messages"],
"results": [{"source": "researcher", "result": str(result["messages"][-1].content)}]
}
async def coder_agent(state: RouterState, runtime: Runtime[Context]) -> Dict[str, Any]:
file_path = os.path.join(base_dir, "souls", "coder", "SOUL.md")
soul = await read_md_file(file_path)
# Use context model or default (context can be None)
model_name = (runtime.context or {}).get("coder_model", DEFAULT_MODEL)
model = ModelAgent(model_name=model_name).load_model()
agent = create_deep_agent(
model=model,
tools=[],
system_prompt=soul,
backend=backend
)
last_msg = state["messages"][-1]
human_msg = HumanMessage(content=last_msg.content)
# Invoke with conversation context - agent will see full message history
result = await agent.ainvoke({"messages": [human_msg]})
result["messages"] = [
m for m in result["messages"] if not isinstance(m, HumanMessage)
]
# Return results with source tracking
return {
"messages": result["messages"],
"results": [{"source": "coder", "result": str(result["messages"][-1].content)}]
}
Ścieżka Programisty przyjmuje parametr backendu, dzięki czemu wykonywanie kodu może odbywać się w izolowanym środowisku, z dala od hosta. Dokumentacja LangChain opisuje kilka opcji backendu, w tym routery kompozytowe, w przewodniku po backendach Deep Agents: https://docs.langchain.com/oss/python/deepagents/backends#compositebackend-router
Logika grafu
Połącz przygotowane elementy w graf stanów:
builder = StateGraph(RouterState, context_schema=Context)
builder.add_node("orchestrator", orchestrator_agent)
builder.add_node("researcher", researcher_agent)
builder.add_node("coder", coder_agent)
builder.add_edge(START, "orchestrator")
# LangGraph API provides persistence automatically
# - langgraph dev: in-memory checkpointer
# - production deploy: PostgreSQL checkpointer (uses POSTGRES_URI from .env)
graph = builder.compile()
Rozwój lokalny może korzystać z mechanizmu checkpointer w pamięci operacyjnej LangGraph. Wdrożenia produkcyjne przekazują URI do Postgresa, dzięki czemu punkty kontrolne przetrwają restarty procesu.
Testowanie
Dwa tryby uruchamiania obejmują cały cykl życiowy: langgraph dev do iteracji lokalnych oraz langgraph up do wykonywania zadań w warunkach produkcyjnych.
Korzystnym testem dymnym jest nadanie nazwy i zapytanie, czy późniejsze etapy wciąż ją pamiętają – co potwierdza zachowanie kontekstu konwersacyjnego zapisanego w punktach kontrolnych. Pytanie o charakterze badawczym powinno zostać skierowane do osoby zajmującej się badaniami i wykorzystać Tavily. Prośba o kodowanie, np. implementację quicksort w C++, powinna trafić do programisty. Gdy wszystkie te trzy ścieżki zadziałają pomyślnie, graf jest gotowy na dodanie kolejnych agentów, narzędzi lub umiejętności w ramach dalszej rozbudowy.
Podsumowanie
Główną zaletą LangGraph pod względem projektowania wieloagentowego jest jego elastyczność. Nic nie wymusza ustalonej liczby agentów ani określonej topologii. Graf może zacząć się od jednego węzła i rozwijać, dodając specjalistycznych agentów, warunkowe ścieżki przekierowania, równoległe gałęzie, pętle, przerwy z udziałem człowieka lub zagłębione podgrafy w miarę wzrastających wymagań.
Literatura pokrewna
- Twoje pierwsze narzędzie – agent AI z LangChain — Stwórz małego agenta matematycznego z LangChain w Pythonie: zainicjuj model do rozmów, zarejestruj narzędzia, pozwól LLM wybrać funkcję do wywołania i zobacz wieloetapowe używanie narzędzi.
- Pauzowanie i wznawianie agentów LangGraph za pomocą interrupt() i Command — Jak funkcje interrupt() i Command w LangGraph wykorzystują punkty kontrolne oraz wątki, aby zapauzować agenta w celu uzyskania akceptacji, edycji lub danych od użytkownika, a następnie bezpiecznie go wznowić później.