Multi-Agent Research Assistant

 question    │    ▼ ┌──────────┐  plan: 3–6 sub-questions (structured) │ planner  │──────────────────────────────────────┐ └──────────┘                                      │ Send(worker, sub_q) × N  (parallel)                                                   ▼                       ┌──────────┐ ┌──────────┐ ┌──────────┐                       │ worker 1 │ │ worker 2 │ │ worker N │   each: search → fetch → read → notes                       └────┬─────┘ └────┬─────┘ └────┬─────┘   budget: steps, tokens, time                            └────────────┼────────────┘                                         ▼  findings (reducer: list append)                                  ┌──────────────┐                                  │ gap check    │── gaps & budget left? ──► another round                                  └──────┬───────┘                                         ▼                                  ┌──────────────┐     ┌──────────────┐                                  │ writer       │ ──► │ reviewer     │ ── issues ──► writer (max 2)                                  └──────────────┘     └──────┬───────┘                                                              ▼                                                    report.md + run_log.json

import operatorfrom typing import Annotated, TypedDictfrom pydantic import BaseModelfrom langgraph.graph import StateGraph, START, ENDfrom langgraph.types import Sendclass SubQuestion(BaseModel):    id: str    question: str    why_it_matters: strclass Plan(BaseModel):    sub_questions: list[SubQuestion]class Note(BaseModel):    id: str    claim: str    url: str    quote: str    sub_question_id: strclass ResearchState(TypedDict):    question: str    plan: list[SubQuestion]    notes: Annotated[list[Note], operator.add]        # parallel workers append safely    failures: Annotated[list[str], operator.add]    rounds: int    report: strclass WorkerInput(TypedDict):    sub: SubQuestiondef planner(state: ResearchState):    plan = planner_llm.with_structured_output(Plan).invoke(        "Break this research question into 3-6 independent, specific sub-questions. "        f"Avoid overlap.\n\nQuestion: {state['question']}")    return {"plan": plan.sub_questions, "rounds": state.get("rounds", 0) + 1}def fan_out(state: ResearchState):    return [Send("worker", {"sub": s}) for s in state["plan"]]def worker(inp: WorkerInput):    try:        notes = run_research_agent(inp["sub"], max_steps=8, max_tokens=40_000, timeout_s=120)        return {"notes": notes}    except Exception as e:        return {"failures": [f"{inp['sub'].id}: {e}"]}def gap_check(state: ResearchState):    covered = {n.sub_question_id for n in state["notes"]}    missing = [s for s in state["plan"] if s.id not in covered]    if missing and state["rounds"] < 2 and within_budget(state):        return [Send("worker", {"sub": s}) for s in missing]    return "writer"def writer(state: ResearchState):    notes = "\n".join(f"[{n.id}] {n.claim} (quote: \"{n.quote}\")" for n in state["notes"])    report = writer_llm.invoke(        f"Question: {state['question']}\nNotes:\n{notes}\nFailed areas: {state['failures']}\n"        "Write a report using only these notes; cite note ids like [n3]; "        "include a 'Contradictions and open questions' section.").content    return {"report": report}g = StateGraph(ResearchState)g.add_node("planner", planner)g.add_node("worker", worker)g.add_node("collect", lambda s: {})g.add_node("writer", writer)g.add_edge(START, "planner")g.add_conditional_edges("planner", fan_out, ["worker"])g.add_edge("worker", "collect")g.add_conditional_edges("collect", gap_check, ["worker", "writer"])g.add_edge("writer", END)          # add reviewer node + loop as a milestonegraph = g.compile()