Build an Enterprise‑Level AI Research Report Generator from Scratch with LangGraph

The article demonstrates how to assemble a complete, production‑grade research‑report generation pipeline that automatically searches multiple information sources, aggregates the data, drafts a report, evaluates its quality, iterates improvements, and finally obtains human approval before publishing.

Core Design Patterns

Fan‑out : Parallel search of five sources reduces total latency to one‑fifth of sequential execution.

Fan‑in : Aggregates the five raw results into a structured analysis.

Evaluator‑Optimizer Loop : Self‑reflection cycle that rewrites the draft until a quality score ≥ 0.85 or a maximum of three iterations.

HITL : Human review after the evaluation passes, ensuring the final draft is vetted.

Checkpointing : SQLite‑based state saver that allows the workflow to resume after interruptions.

Langfuse Observability : End‑to‑end tracing of each node for performance and cost analysis.

State Design

from typing import TypedDict, Annotated, Optional
import operator

class ResearchReportState(TypedDict):
    # ── Input ──────────────────────────────────────
    topic: str  # research topic
    date: str   # report date
    target_audience: str  # influences writing style

    # ── Parallel search results (Reducer: append) ──────────────
    raw_sources: Annotated[list[dict], operator.add]

    # ── Analysis & Writing (cover mode) ─────────────────────
    structured_analysis: str  # structured conclusions
    report_draft: str          # draft report
    improvement_notes: str    # notes written by the evaluator

    # ── Quality control ──────────────────────────────────
    quality_score: float      # 0‑1 normalized score
    quality_feedback: str     # detailed reviewer comments
    iteration_count: int      # prevents infinite loops

    # ── Human operation records ──────────────────────────────
    human_decision: Optional[str]   # "approve" / "reject"
    human_feedback: Optional[str]
    human_decision_time: Optional[str]

    # ── Final output ──────────────────────────────────
    final_report: str
    publish_url: Optional[str]

Key points include using operator.add as a reducer so that each parallel node appends its result without overwriting, keeping the report_draft in overwrite mode, and limiting the iteration_count to three to avoid endless loops.

Step‑by‑Step Nodes

1️⃣ Parallel Search Nodes

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END

llm_fast = ChatOpenAI(model="gpt-4o-mini", temperature=0)   # cheap search model
llm_good = ChatOpenAI(model="gpt-4o", temperature=0.3)       # better writing model
llm_strict = ChatOpenAI(model="gpt-4o", temperature=0)    # strict evaluator model

def search_industry_news(state: ResearchReportState) -> dict:
    """Search industry news"""
    results = web_search(query=f"{state['topic']} 行业新闻 最新动态", n_results=5)
    return {"raw_sources": [{"type": "industry_news", "data": results}]}

# Similar functions for academic, competitor, financial, and social searches …

Each function returns a dictionary with a raw_sources entry; the reducer concatenates these entries.

2️⃣ Aggregation (Fan‑in) Node

def aggregate_node(state: ResearchReportState) -> dict:
    """Combine the five raw sources into a structured analysis"""
    sources_by_type = {}
    for source in state["raw_sources"]:
        t = source["type"]
        sources_by_type.setdefault(t, []).append(source["data"])
    formatted = f"""
## 行业新闻
{format_sources(sources_by_type.get('industry_news', []))}

## 学术研究
{format_sources(sources_by_type.get('academic', []))}

## 竞品动态
{format_sources(sources_by_type.get('competitor', []))}

## 财务数据
{format_sources(sources_by_type.get('financial', []))}

## 市场讨论
{format_sources(sources_by_type.get('social', []))}
"""
    prompt = f"""
研究主题：{state['topic']}
目标读者：{state['target_audience']}
基于以下多源数据，生成结构化分析：
{formatted}
分析框架：
1. 核心趋势（3‑5 条，每条有数据支撑）
2. 主要机会（2‑3 个，有具体依据）
3. 潜在风险（2‑3 个，有量化评估）
4. 竞争格局（关键玩家和市场份额）
5. 数据亮点（最有价值的 3 个数据点）
"""
    result = llm_good.invoke(prompt)
    return {"structured_analysis": result.content}

The node first groups results by type, formats them, and then asks the LLM to produce a concise, data‑backed analysis.

3️⃣ Writing Node

def write_node(state: ResearchReportState) -> dict:
    """Draft the report based on the structured analysis"""
    improvement_context = ""
    if state.get("improvement_notes"):
        improvement_context = f"""【上一版本的改进意见，请针对性修改】
{state['improvement_notes']}"""
    prompt = f"""
研究主题：{state['topic']}
报告日期：{state['date']}
目标读者：{state['target_audience']}

结构化分析：
{state['structured_analysis']}

{improvement_context}

请起草一份专业研报，要求：
- 总字数 1500‑2000 字
- 结构：执行摘要（200 字）+ 市场概况 + 竞争分析 + 机会与风险 + 结论与建议
- 每个观点都有数据支撑
- 语言专业但不晦涩，适合 {state['target_audience']} 阅读
- 如有改进意见，必须针对性修改
"""
    result = llm_good.invoke(prompt)
    return {"report_draft": result.content,
            "iteration_count": state.get("iteration_count", 0) + 1}

If the evaluator returned improvement notes, they are injected into the prompt so the next draft addresses specific problems.

4️⃣ Evaluation Node (Evaluator‑Optimizer Core)

def evaluate_node(state: ResearchReportState) -> dict:
    """Score the draft and produce concrete improvement suggestions"""
    prompt = f"""
你是一个严格的研报质量审核专家。

请评估以下研报的质量（0‑100 分）：
【研报内容】
{state['report_draft']}

【评估维度】
1. 数据支撑（30 分）
2. 逻辑严密（25 分）
3. 实用价值（25 分）
4. 写作质量（20 分）

【输出格式】
总分：XX 分
各维度得分：数据支撑XX/30，逻辑严密XX/25，实用价值XX/25，写作质量XX/20
主要问题：
1. [具体问题1，指出在哪里、怎么改]
2. [具体问题2，指出在哪里、怎么改]
改进建议：[针对主要问题的具体修改方向]
"""
    result = llm_strict.invoke(prompt)
    score = parse_quality_score(result.content)  # extracts the total score
    feedback = result.content
    improvement_notes = extract_improvement_notes(result.content)
    return {"quality_score": score / 100,
            "quality_feedback": feedback,
            "improvement_notes": improvement_notes}

def route_after_evaluation(state: ResearchReportState) -> str:
    score = state.get('quality_score', 0)
    iterations = state.get('iteration_count', 0)
    if score >= 0.85:
        return "human_review"
    if iterations >= 3:
        return "human_review"
    return "improve"

The evaluator uses a zero‑temperature model to guarantee deterministic scoring and asks for pinpointed issues, which are later turned into actionable improvement instructions.

5️⃣ Improvement Node

def improve_node(state: ResearchReportState) -> dict:
    """Translate evaluator feedback into concrete rewrite instructions"""
    prompt = f"""
当前报告质量评分：{state['quality_score'] * 100:.0f} 分
评估意见：
{state['quality_feedback']}

请生成一份简洁的改进指令（3‑5 条），每条指令要：
- 指出具体问题所在
- 说明如何修改
- 给出修改示例（如果可能）
格式：
1. [问题位置]：[具体修改方法]
"""
    result = llm_fast.invoke(prompt)
    return {"improvement_notes": result.content}

The node does not edit the draft directly; it only produces a set of clear instructions that the writing node will consume.

6️⃣ HITL (Human‑in‑the‑Loop) Node

def hitl_node(state: ResearchReportState) -> dict:
    """Placeholder for manual review; actual pause is handled by LangGraph's interrupt_before"""
    if state.get("human_decision") == "approve":
        return {}
    if state.get("human_decision") == "reject":
        return {"report_draft": "",
                "quality_score": 0,
                "iteration_count": 0,
                "improvement_notes": state.get("human_feedback", "")}
    return {}

def route_after_hitl(state: ResearchReportState) -> str:
    decision = state.get("human_decision", "")
    if decision == "approve":
        return "publish"
    if decision == "reject":
        return "write"
    return "publish"

The node itself does nothing; the workflow pauses before it, waiting for a human to set human_decision and optional feedback.

7️⃣ Publish Node

def publish_node(state: ResearchReportState) -> dict:
    """Output the final report and store it"""
    final_report = state.get("human_modified_draft") or state["report_draft"]
    publish_url = publish_report(content=final_report,
                                 topic=state["topic"],
                                 date=state["date"])
    return {"final_report": final_report,
            "publish_url": publish_url}

The function either uses the human‑edited draft or the AI‑generated one and then calls a placeholder publish_report to store or distribute the document.

Graph Assembly

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver
from langfuse.callback import CallbackHandler
import os

builder = StateGraph(ResearchReportState)
# add all nodes … (search_* , aggregate , write , evaluate , improve , hitl , publish)
builder.set_entry_point("trigger")
# trigger node fans out to the five search nodes (see source for code)
# edges connect search → aggregate → write → evaluate → conditional routing → improve/write or hitl → conditional routing → publish → END
checkpointer = SqliteSaver.from_conn_string("research_reports.db")
app = builder.compile(checkpointer=checkpointer,
                     interrupt_before=["hitl"])

The graph uses a dedicated trigger node because LangGraph can only have a single entry point; the trigger fans out to the parallel search nodes, achieving true fan‑out.

Observability with Langfuse

By wrapping each search function with langfuse.trace, the dashboard shows per‑node latency, token consumption, and output size, making it easy to spot bottlenecks (e.g., academic search may be slower) and costly nodes (writing and evaluation typically consume the most tokens).

Checkpointing & Resumability

Using SqliteSaver ensures that if the workflow is interrupted—whether by a crash or a manual pause—the state is persisted. The resume_report function demonstrates how to retrieve the saved state and continue execution from the last node.

Pre‑Run Checklist

All parallel nodes use operator.add as a reducer to avoid data loss.

Loop nodes use overwrite mode because history is not required.

State does not contain unnecessary large fields, keeping context size manageable.

Evaluation loop has both a quality‑score threshold (0.85) and a maximum iteration count (3) to prevent infinite runs.

Routing functions cover every possible branch.

HITL pause is correctly configured via interrupt_before.

Model tiering (cheap for search, medium for writing, strict for evaluation) controls cost.

Langfuse traces are added to critical nodes for debugging and performance analysis.

Key Takeaways

Fan‑out/Fan‑in enables parallel data collection and deterministic aggregation.

Evaluator‑Optimizer loop improves quality through concrete, scored feedback and a hard iteration cap.

HITL provides a safety net before publishing, with the workflow automatically pausing for human input.

Reducer design (using operator.add) is essential for preserving all parallel results.

Model tiering balances cost and performance across the pipeline.

Checkpointing + Langfuse makes the system production‑ready by guaranteeing resumability and full observability.