Организуйте работу исследователей и программистов с помощью LangGraph

Создание одного ИИ-агента — это простая задача по сравнению с координацией нескольких специалистов, которые знают свои обязанности и моменты передачи работы. Рабочий процесс LangGraph позволяет назначить Оркестратора для управления маршрутизацией, в то время как Исследователь и Программист занимаются конкретными задачами. Каждый агент сосредоточен на одной функции — поиске информации, реализации или координации — так что вса структура совместно работает над получением одного ответа, вместо того чтобы заставлять одну модель выполнять все этапы.

Разработка

Начало нового проекта

При наличии Python создайте структуру проекта LangGraph:

Langgraph new

Эта команда создает повторно используемый шаблон. Добавьте зависимости для выполнения с помощью 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"

Настройте файл .env для отслеживания действий LangSmith, URI базы данных Postgres в качестве точки контроля в производственной среде, а также ключ Tavily для поиска в интернете:

# 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 разделяет следы по проектам, Postgres хранит сообщения в состоянии чекпоинта для надежной работы, а Tavily обеспечивает поддержку инструмента поиска исследователя.

Агент SOUL

Каждому агенту требуется системное указание («душа»), сохраненное в формате Markdown. Организуйте их в папке src/agent:

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

Создайте файлы SOUL.md для оркестратора, исследователя и программиста. Пример содержимого указания:

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

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

Модель Ollama

В примере используется облачная модель Gemma 4, размещенная на платформе Ollama, поэтому не требуется большое локальное скачивание. Для экспериментов можно использовать других поставщиков.

ollama pull gemma4:cloud

Определите небольшой обертку для модели в папке 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")

Обертка скрывает детали поставщика от графа: узлы вызывают один класс и передают необходимое имя модели.

Глобальная переменная

Инициализируйте общие настройки перед объявлением узлов или инструментов:

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

Исследователю требуется поиск в интернете. Tavily реализован как инструмент, создаваемый по мере необходимости, и он считывает значение 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

Передача контроля

Передача контроля — это инструмент, возвращающий команду графа для передачи управления другому агенту. Обычные инструменты выполняют работу; передача контроля определяет, кто будет выполнять следующий шаг. Она важна для:

1. Координации множества агентов
2. Агентов с разными наборами инструментов
3. Оркестратора, который делегирует задачи вместо того, чтобы выполнять их сам

Определите механизмы передачи контроля для исследователей и программистов:

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

Узлы агентов

Реализуйте три узла — Оркестратор, Исследователь и Кодер — часто с разными именами моделей, чтобы можно было настраивать затраты и точность в зависимости от роли:

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

Путь Кодера принимает параметр backend, что позволяет выполнять код в изолированной среде, отделенной от хоста. В руководстве LangChain по backend для Deep Agents описано несколько вариантов backend, включая комбинированные маршрутизаторы: https://docs.langchain.com/oss/python/deepagents/backends#compositebackend-router

Логика графа

Соедините подготовленные элементы в граф состояний:

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

Для локальной разработки можно использовать встроенный в LangGraph механизм сохранения состояния в памяти. При развертывании в производстве передается URI базы данных Postgres, что позволяет сохранять состояния после перезапуска процесса.

Тестирование

Для управления жизненным циклом существуют два режима запуска: langgraph dev для локальных итераций и langgraph up для работы в режиме производства.

Полезным тестом является присвоение имени и проверка на то, помнят ли участники его позже — это подтверждает сохранение контекста диалога. Вопросы, связанные с исследованиями, должны направляться к специалисту по исследованиям для тестирования Tavily. Запросы на программирование, такие как реализация алгоритма быстрого сортирования на C++, следует передавать программисту. Если все три варианта работы успешны, граф готов к добавлению дополнительных агентов, инструментов или навыков в рамках дальнейшего расширения.

Краткое резюме

Главное преимущество LangGraph — его гибкость в проектировании многоагентных систем. Нет никаких ограничений по фиксированному количеству агентов или определенной топологии. Граф может начинаться с одного узла и по мере усложнения требований развиваться с помощью специализированных агентов, условного маршрутизации, параллельных ветвей, циклов, пауз для участия человека или вложенных подграфов.