Multi-agent research assistant using LangGraph and n4n.ai

Decomposing a complex research question into subtasks and coordinating specialized agents is the fastest way to get reliable answers from LLMs. This guide walks through building a langgraph n4n.ai multi-agent research assistant that splits queries, runs parallel retrieval and analysis, and merges findings into a cited report.

Step 1: Install dependencies and scaffold the project

Create a virtual environment and install the minimal stack. You need LangGraph for orchestration, the OpenAI-compatible LangChain wrapper for model calls, and dotenv for local secrets.

python -m venv .venv
source .venv/bin/activate
pip install langgraph langchain-openai python-dotenv

Create a .env file with your gateway key. The endpoint is OpenAI-compatible, so any tooling expecting OPENAI_API_KEY can be repointed with one environment variable.

echo "N4N_API_KEY=sk-your-key-here" > .env

Keep the key out of source control. LangGraph compiles to a static graph; secrets belong in the runtime environment, not in node definitions.

Step 2: Configure the model client

n4n.ai provides an OpenAI-compatible endpoint that addresses 240+ models with automatic fallback when a provider is degraded. Point ChatOpenAI at the base URL and select a default model. For a research assistant, a mid-tier model handles decomposition well, while synthesis benefits from a stronger one.

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()

default_llm = ChatOpenAI(
    model="gpt-4o-mini",
    api_key=os.environ["N4N_API_KEY"],
    base_url="https://api.n4n.ai/v1",
    temperature=0.2,
)

strong_llm = ChatOpenAI(
    model="claude-3-5-sonnet",
    api_key=os.environ["N4N_API_KEY"],
    base_url="https://api.n4n.ai/v1",
    temperature=0.1,
)

The gateway honors client routing directives and forwards provider cache-control hints. If you pin a model that is temporarily rate-limited, the request fails over without code changes. Per-token usage metering is returned in the response metadata, so you can log cost per agent turn.

Step 3: Define the shared state contract

LangGraph flows data through a typed state object. Design it to carry the original query, decomposed subqueries, raw research, critique, and the final report. Use immutable updates—each node returns a partial dict that is merged.

from typing import TypedDict, List, Dict

class ResearchState(TypedDict):
    query: str
    subqueries: List[str]
    raw_results: Dict[str, str]
    critique: str
    final_report: str

Avoid stuffing transient scratch data into the state. If a node needs a temporary variable, compute it locally and return only what downstream nodes require.

Step 4: Implement the agent nodes

Each node is a plain function taking the state and returning a partial update. Start with decomposition using the cheaper model.

from langchain_core.messages import HumanMessage

def decompose(state: ResearchState) -> dict:
    prompt = f"Break this research question into 3-5 distinct subqueries.\nQuery: {state['query']}\nReturn one per line."
    resp = default_llm.invoke([HumanMessage(content=prompt)])
    subs = [line.strip("- ").strip() for line in resp.content.split("\n") if line.strip()]
    return {"subqueries": subs}

The research node iterates subqueries. In production you would parallelize with Send or asyncio, but a sequential loop is clearer for a first build.

def research(state: ResearchState) -> dict:
    results = {}
    for q in state["subqueries"]:
        prompt = f"Research the following concisely with factual claims:\n{q}"
        resp = default_llm.invoke([HumanMessage(content=prompt)])
        results[q] = resp.content
    return {"raw_results": results}

Critique checks for gaps before synthesis. Use the stronger model here because it needs to reason about coverage.

def critique(state: ResearchState) -> dict:
    joined = "\n".join(f"Subquery: {k}\nFindings: {v}" for k, v in state["raw_results"].items())
    prompt = f"Identify missing perspectives or weak evidence in this research:\n{joined}"
    resp = strong_llm.invoke([HumanMessage(content=prompt)])
    return {"critique": resp.content}

Synthesize produces the final artifact.

def synthesize(state: ResearchState) -> dict:
    joined = "\n".join(f"Subquery: {k}\nFindings: {v}" for k, v in state["raw_results"].items())
    prompt = f"Write a structured report. Use the research and address the critique.\nResearch:\n{joined}\nCritique:\n{state['critique']}"
    resp = strong_llm.invoke([HumanMessage(content=prompt)])
    return {"final_report": resp.content}

Step 5: Wire the graph

Compose the nodes into a linear pipeline with an entry point and terminal edge. LangGraph validates state transitions at compile time.

from langgraph.graph import StateGraph, END

workflow = StateGraph(ResearchState)
workflow.add_node("decompose", decompose)
workflow.add_node("research", research)
workflow.add_node("critique", critique)
workflow.add_node("synthesize", synthesize)

workflow.set_entry_point("decompose")
workflow.add_edge("decompose", "research")
workflow.add_edge("research", "critique")
workflow.add_edge("critique", "synthesize")
workflow.add_edge("synthesize", END)

app = workflow.compile()

For branching logic—say, repeating research if critique flags major gaps—replace the edge with add_conditional_edges and a router function. Keep routers pure and fast; they should not call the LLM.

Step 6: Execute and verify

Invoke with a real query and inspect the output shape.

result = app.invoke({
    "query": "What are the trade-offs between WebAssembly and containers for edge deployment?"
})

assert result["subqueries"], "Decomposition failed"
assert len(result["raw_results"]) == len(result["subqueries"]), "Research mismatch"
assert len(result["final_report"]) > 200, "Report too short"
print(result["final_report"])

Verification succeeds when subqueries is non-empty, raw_results has a key for every subquery, and final_report is a coherent multi-paragraph answer that references the critique. Run it twice to confirm the gateway’s fallback does not alter the graph contract when a model is swapped underneath.

Step 7: Harden for production

The langgraph n4n.ai multi-agent research assistant above is a skeleton. Three changes make it ship-ready.

First, isolate model configuration per node via closure or class, so you can swap default_llm for a local model on the same endpoint without touching graph topology. Second, add a timeout wrapper around llm.invoke—the gateway fails over providers, but your client should not hang indefinitely. Third, emit usage_metadata to your logging sink after each node to track per-agent token spend.

def instrumented_research(state: ResearchState) -> dict:
    results = {}
    for q in state["subqueries"]:
        resp = default_llm.invoke([HumanMessage(content=f"Research: {q}")])
        if hasattr(resp, "usage_metadata"):
            print(f"research token usage: {resp.usage_metadata}")
        results[q] = resp.content
    return {"raw_results": results}

LangGraph’s checkpointing integrates cleanly if you need to resume long-running research after a crash. Pass a MemorySaver or Postgres-backed saver to compile().

Multi-agent systems earn their complexity only when single prompts degrade. Start with this pipeline, measure where answers break, and add conditional loops or specialist agents only where the data demands it.