Orchestrieren Sie Forscher- und Programmierer-Agenten mit LangGraph

Der Aufbau eines einzelnen KI-Agenten ist im Vergleich zur Koordination mehrerer Spezialisten, die ihre Rollen sowie den richtigen Zeitpunkt für die Aufgabenübertragung kennen, unkompliziert. Ein LangGraph-Arbeitsablauf kann einen Orchestrator für die Routensteuerung einsetzen, während ein Researcher und ein Coder spezifische Aufgaben bearbeiten. Jeder Agent konzentriert sich auf eine bestimmte Aufgabe – Abfragen, Implementierung oder Koordination – sodass das Gesamtsystem gemeinsam zu einer Antwort gelangt, anstatt ein einzelnes Modell alle Schritte ausführen zu müssen.

Entwicklung

N neues Projekt starten

Mit Python verfügbar, erstellen Sie eine Grundstruktur für ein LangGraph-Projekt:

Langgraph new

Dieser Befehl legt ein wiederverwendbares Template ab. Fügen Sie die Laufzeitabhängigkeiten mit uv hinzu:

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"

Konfigurieren Sie die Datei .env für das Tracking durch LangSmith, einen Postgres-URI als Referenz für die Produktivumgebung sowie eine Tavily-Schlüssel für Suchanfragen im Web:

# 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 trennt die Spuren nach Projekt, Postgres speichert die für ausfallsichere Ausführungen markierten Nachrichten, und Tavily unterstützt das Suchwerkzeug des Forschers.

Agent SOUL

Jeder Agent benötigt eine Systemanweisung (“Soul”), die als Markdown gespeichert wird. Organisieren Sie sie unter src/agent:

src/agent/
---------- souls/
----------------- orchestrator/
----------------- researcher/
----------------- coder/

Erstellen Sie SOUL.md-Dateien für den Orchestrierer, den Forscher und den Programmierer. Beispiele für Anweisungstexte:

# 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.

Diese Dateien kodieren Rolle, Aufgabe sowie Betriebsbeschränkungen, sodass jeder Knoten eine stabile Persönlichkeit lädt, anstatt lange Zeichenketten in Python zu verwenden.

Ollama-Modell

Die Anleitung verwendet ein auf Ollama gehostetes Gemma 4-Cloudmodell, wodurch kein großer lokaler Download erforderlich ist. Andere Anbieter können für Experimente verwendet werden.

ollama pull gemma4:cloud

Definieren Sie einen kleinen Modell-Wrapper unter 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")

Der Wrapper hält die Details des Anbieters außerhalb des Graphen: Die Knoten rufen eine Klasse auf und übergeben den Namen des Modells, das sie benötigen.

Globale Variable

Initialisieren Sie die gemeinsamen Standardwerte vor der Deklaration von Knoten oder Tools:

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)

Tavily-Tool

Der Forscher benötigt eine Internetsuche. Tavily ist als verzögert erstelltes Tool konfiguriert, das TAVILY_API_KEY liest:

# 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

Übergaben

Eine Übergabe ist ein Tool, das einen Graphen-Befehl zurückgibt und die Steuerung an einen anderen Agenten überträgt. Reguläre Tools führen Aufgaben aus; Übergaben wechseln den Verantwortlichen für den nächsten Schritt. Sie sind wichtig für:

1. Mehr-Agenten-Koordination
2. Agenten mit unterschiedlichen Toolsets
3. Einen Orchestrierer, der Aufgaben delegiert anstatt sie selbst auszuführen

Definieren Sie Übergaben für Forscher und Programmierer:

# 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,
    )

Agentenknoten

Implementieren Sie drei Knoten – Orchestrator, Researcher und Coder – oft mit unterschiedlichen Modellnamen, damit Kosten und Genauigkeit je nach Rolle angepasst werden können:

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)}]
    }

Der Coder-Pfad akzeptiert einen Backend-Parameter, sodass die Codeausführung in einer isolierten Umgebung fernab des Hosts stattfinden kann. LangChain dokumentiert mehrere Backend-Optionen, einschließlich komplexer Router, in der Anleitung zu Deep Agents Backends: https://docs.langchain.com/oss/python/deepagents/backends#compositebackend-router

Graphenlogik

Verschließen Sie die vorbereiteten Komponenten zu einem Zustandsgraphen:

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()

Bei der lokalen Entwicklung kann auf den in-Memory-Checkpointer von LangGraph zurückgegriffen werden. Bei Produktivumgebungen wird ein Postgres-URI übergeben, damit die Checkpoints nach einem Neustart des Prozesses erhalten bleiben.

Testing

Zwei Startmodi decken den gesamten Lebenszyklus ab: langgraph dev für lokale Iterationen und langgraph up für produktionsnahe Ausführungen.

Ein nützlicher Smoke-Test besteht darin, einen Namen zu geben und zu prüfen, ob die späteren Schritte sich noch daran erinnern – dadurch wird der gespeicherte konversationelle Kontext bestätigt. Eine für Forschungszwecke formulierte Frage sollte an den Forscher weitergeleitet werden, um Tavily zu testen. Eine Programmieranfrage wie die Implementierung von QuickSort in C++ sollte an den Programmierer weitergeleitet werden. Wenn diese drei Ansätze erfolgreich sind, ist das Graphenmodell bereit für weitere Agenten, Tools oder Fähigkeiten in einer anschließenden Erweiterung.

Zusammenfassung

Die Flexibilität von LangGraph ist der wichtigste Vorteil für das Design mehrerer Agenten. Nichts erzwingt eine feste Anzahl an Agenten oder eine bestimmte Topologie. Ein Graph kann mit einem Knoten beginnen und im Laufe der Zeit zu spezialisierten Agenten, bedingtem Routing, parallelen Verzweigungen, Schleifen, Pausen mit menschlicher Beteiligung oder verschachtelten Untergraphen auswachsen, je nachdem wie die Anforderungen konkretisieren.