Multi-Agent BI Pipeline with LangGraph

Orchestrate 4 AI agents to research and generate intelligence reports.

Introduction

30 Second Summary

Every company needs research to make smart decisions. Gathering data from multiple sources, analyzing patterns, and producing actionable recommendations takes hours when done manually.

In this project, you will build a multi-agent Business Intelligence Pipeline that orchestrates four specialized AI agents through a LangGraph StateGraph to produce structured intelligence reports from natural language business questions. You'll implement production-grade patterns including guardrails, token optimization, cost tracking, and a human-in-the-loop approval gate.

What You'll Build

A multi-agent pipeline that decomposes business questions into research tasks, searches the web, analyzes findings, and generates a structured intelligence brief with full observability and human oversight.

By the end of this project, you'll have:

  • A LangGraph StateGraph orchestrating 4 agent nodes (Planner, Researcher, Analyst, Report Writer) with conditional routing and a human approval gate.
  • Production-grade guardrails with input validation, output validation, a custom exception hierarchy, and per-agent token/cost tracking.
  • Full LangSmith tracing providing observability into every agent decision, tool call, and state transition.
  • Secret Mission: An automated evaluation harness that scores report quality using LLM-as-judge and logs metrics to LangSmith.

Are there any prerequisites?

This is a hard-difficulty project. You'll need Python 3.10+ installed, an OpenAI API key, and familiarity with Python basics. You'll sign up for Tavily (free) and LangSmith (free) during Step 1.

Set Up Your Environment

A multi-agent pipeline has several moving parts: an orchestration framework, an LLM provider, a web search tool, and an observability platform. Before writing any agent logic, you need a clean Python environment with all of these wired up and verified.

In this step, you will create a project directory, install the required packages, configure API keys for three services, and run a quick connectivity test to confirm everything works together.

In this step, get ready to:
  • Set up a Python project with a virtual environment and install all dependencies.
  • Configure API keys for OpenAI, Tavily, and LangSmith.
  • Create the project file structure and verify connectivity with a smoke test.
Create your project and install dependencies

Every Python project starts with an isolated virtual environment. This prevents package version conflicts with other projects on your machine.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal and press Enter to open it.

First, confirm your Python version meets the minimum requirement of 3.10.

  • Check your Python version by running this command:
python3 --version

✔️ I see version 3.10 or higher

Your Python version is ready. Continue to the next instruction below.

ⓧ I see a version lower than 3.10

You need Python 3.10+ for this project. The packages we install require it.

  • Visit python.org/downloads and download Python 3.10 or newer.
  • Run the installer and follow the prompts.
  • On macOS, the system Python may still resolve to an older version. Use python3.10 or python3.11 explicitly, or use a version manager like pyenv to set the active version.
  • After installing, re-run python3 --version to confirm it reports 3.10 or higher.

ⓧ Command not found

Python is not installed on your system.

  • Visit python.org/downloads and download Python 3.10 or newer.
  • Run the installer. On Windows, check the box that says Add Python to PATH before clicking Install.
  • Close and reopen your terminal, then run python3 --version again.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder, move into it, and set up a virtual environment by running these commands:
mkdir bi-pipeline
cd bi-pipeline
python3 -m venv venv
  • Activate the virtual environment by running this command:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

You should see (venv) appear at the beginning of your terminal prompt.

  • Install all required packages by running this command:
pip install langgraph==1.2.6 langchain-openai==1.3.3 langchain-tavily==0.2.18 langsmith pydantic python-dotenv

What are these packages?

  • langgraph is the multi-agent orchestration framework that manages your pipeline's state and routing.
  • langchain-openai connects your agents to GPT-4o.
  • langchain-tavily gives your agents web search capabilities.
  • langsmith provides tracing and observability for every agent decision.
  • pydantic validates your pipeline's state at each step.
  • python-dotenv loads environment variables from your .env file.

The install takes about a minute. You should see a success message ending with Successfully installed followed by a list of packages.

Seeing installation errors?

  • Make sure (venv) appears in your terminal prompt. If it does not, re-run the activate command above.
  • If you see a permissions error, do not use sudo. Deactivate, delete the venv folder, and recreate it.
  • help me debug a pip install error in my virtual environment
Get your API keys and create the .env file

Your pipeline needs three API keys: one for OpenAI (which you already have), one for Tavily (web search), and one for LangSmith (observability).

  • Open https://app.tavily.com in your browser.
  • Click Sign up and create an account using Google, GitHub, or email.
  • Once logged in, copy your API key from the dashboard.

What is Tavily?

Tavily is a search API purpose-built for LLM agents. It returns results optimized for AI consumption rather than raw HTML. You get 1,000 free API credits every month with no credit card required.

  • Open https://smith.langchain.com in your browser.
  • Click Sign Up and choose the US data region.
  • Create an account using Google, GitHub, or email.
  • After signing in, click Settings in the left sidebar.
  • Select API Keys, then click Create API Key.
  • Copy the key immediately. LangSmith only shows it once.

What is LangSmith?

LangSmith is an observability platform for LLM applications. It records every agent call, tool invocation, and decision as a trace you can inspect visually. The free Developer plan includes 5,000 traces per month.

Now create a .env file to store all three keys securely.

  • Switch back to Cursor. Open the bi-pipeline folder (the one you just created on your Desktop).
  • Create a new file called .env in the root of bi-pipeline and paste the following:
OPENAI_API_KEY=your_openai_key
TAVILY_API_KEY=your_tavily_key
LANGSMITH_API_KEY=your_langsmith_key
LANGSMITH_TRACING=true
LANGSMITH_PROJECT=business-intel-pipeline
  • Replace your_openai_key with your actual OpenAI API key.
  • Replace your_tavily_key with the Tavily key you just copied.
  • Replace your_langsmith_key with the LangSmith key you just copied.
  • Save the file.

Keep your .env file private

Never commit .env to version control. If you add a .gitignore later, include .env in it.

Create the project file structure and verify connectivity

Your pipeline code will be organized into modules. The agents/ directory holds agent logic, and core/ holds shared infrastructure like logging, guardrails, and state definitions.

  • Create the project directories and files by running these commands in your terminal (make sure you are inside bi-pipeline):
mkdir agents core
touch main.py config.py agents/__init__.py core/__init__.py
  • Confirm the files exist by running:
ls

You should see agents, core, main.py, config.py, .env, and venv in the output.

Now write the configuration module that loads your keys and defines project constants.

  • Open config.py in Cursor and paste the following code:
import os
from dotenv import load_dotenv

load_dotenv()

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
TAVILY_API_KEY = os.getenv("TAVILY_API_KEY")
LANGSMITH_API_KEY = os.getenv("LANGSMITH_API_KEY")

os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_PROJECT"] = "business-intel-pipeline"
os.environ["LANGSMITH_API_KEY"] = LANGSMITH_API_KEY or ""
os.environ["OPENAI_API_KEY"] = OPENAI_API_KEY or ""
os.environ["TAVILY_API_KEY"] = TAVILY_API_KEY or ""

MODEL_NAME = "gpt-4o"
INPUT_PRICE_PER_1M = 2.50
OUTPUT_PRICE_PER_1M = 10.00
MAX_RESEARCH_LOOPS = 2
QUALITY_THRESHOLD = 6

What does this code do?

  • load_dotenv() reads your .env file and loads the keys into environment variables.
  • The os.environ lines ensure the keys are available to LangSmith, OpenAI, and Tavily SDKs throughout your app.
  • MODEL_NAME sets GPT-4o as the model all agents will use.
  • INPUT_PRICE_PER_1M and OUTPUT_PRICE_PER_1M store GPT-4o's pricing for cost tracking later.
  • MAX_RESEARCH_LOOPS and QUALITY_THRESHOLD control the conditional routing logic you will build in Step 4.
  • Save config.py.

Now verify that all three services are reachable.

  • Run this smoke test in your terminal to confirm connectivity to OpenAI, Tavily, and LangSmith:
python3 -c "
import config
from langchain_openai import ChatOpenAI
from langchain_tavily import TavilySearch
from langsmith import Client

llm = ChatOpenAI(model=config.MODEL_NAME)
response = llm.invoke('Say hello in one word.')
print(f'OpenAI: {response.content}')

search = TavilySearch(max_results=1)
result = search.invoke({'query': 'test query'})
print(f'Tavily: connected')

client = Client()
print(f'LangSmith: connected')
print('All services verified!')
"

You should see output confirming each service connected successfully, ending with All services verified!.

Seeing an authentication or connection error?

  • Double-check your .env file for typos. Each key should be on its own line with no spaces around the = sign.
  • Make sure you saved .env before running the test.
  • If you see a Tavily error, confirm your key starts with tvly-.
  • help me debug API connectivity issues with OpenAI, Tavily, or LangSmith

✔️ Awesome, I've got everything!

Great. Double check you have saved both .env and config.py.

ⓧ I'd like to double check the full code

Here is the complete config.py file:

import os
from dotenv import load_dotenv

load_dotenv()

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
TAVILY_API_KEY = os.getenv("TAVILY_API_KEY")
LANGSMITH_API_KEY = os.getenv("LANGSMITH_API_KEY")

os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_PROJECT"] = "business-intel-pipeline"
os.environ["LANGSMITH_API_KEY"] = LANGSMITH_API_KEY or ""
os.environ["OPENAI_API_KEY"] = OPENAI_API_KEY or ""
os.environ["TAVILY_API_KEY"] = TAVILY_API_KEY or ""

MODEL_NAME = "gpt-4o"
INPUT_PRICE_PER_1M = 2.50
OUTPUT_PRICE_PER_1M = 10.00
MAX_RESEARCH_LOOPS = 2
QUALITY_THRESHOLD = 6

And here is the complete .env file:

OPENAI_API_KEY=your_openai_key
TAVILY_API_KEY=your_tavily_key
LANGSMITH_API_KEY=your_langsmith_key
LANGSMITH_TRACING=true
LANGSMITH_PROJECT=business-intel-pipeline

Your environment is fully configured. Next up, you will build the production foundation with state management, custom exceptions, structured logging, and guardrails.

Build the Production Foundation

Your environment is configured and all services are connected. Now it's time to build the production infrastructure that separates a toy script from a deployable system.

In this step, you'll create the shared state model that flows between agents, a custom exception hierarchy for graceful error handling, structured JSON logging for observability, and input/output guardrails that protect the pipeline from bad data.

In this step, get ready to:
  • Define a typed pipeline state and custom exception classes.
  • Build structured JSON logging and a token cost tracker.
  • Build input and output guardrails that validate data before and after processing.
Define the pipeline state and custom exceptions

Every agent in the pipeline reads from and writes to a shared state object. Defining this state upfront gives you a single contract that every node must follow.

  • In Cursor, create a new file called core/state.py inside your bi-pipeline project folder.
  • Add the pipeline state model by pasting this code:
from typing import Annotated, TypedDict
import operator


class PipelineState(TypedDict):
    query: str
    research_plan: list[str]
    raw_findings: str
    compressed_findings: str
    analysis: str
    quality_score: int
    report: str
    approval_status: str
    research_loops: int
    token_usage: dict[str, dict[str, int]]
    errors: Annotated[list[str], operator.add]

What does this code do?

PipelineState is a TypedDict that defines every piece of data that flows between agents. Each agent reads what it needs and writes its output back to the state.

The errors field uses Annotated[list[str], operator.add] which tells LangGraph to accumulate errors across nodes by concatenating lists rather than overwriting. This means every agent can append errors without losing messages from previous agents.

  • Save core/state.py.
  • Verify the import works by running this command in your terminal:
python -c "from core.state import PipelineState; print('PipelineState loaded OK')"
  • You should see PipelineState loaded OK printed in the terminal.

Seeing a ModuleNotFoundError?

Make sure you are running the command from your bi-pipeline project folder (the one containing the core/ directory). If you're in a different directory, cd back to it first.

Also check that your virtual environment is still active. You should see (venv) at the start of your terminal prompt.

Help me debug my ModuleNotFoundError for core.state

✔️ Awesome, I've got everything!

Great. Make sure you've saved the file before moving on.

ⓧ I'd like to double check the full code

from typing import Annotated, TypedDict
import operator


class PipelineState(TypedDict):
    query: str
    research_plan: list[str]
    raw_findings: str
    compressed_findings: str
    analysis: str
    quality_score: int
    report: str
    approval_status: str
    research_loops: int
    token_usage: dict[str, dict[str, int]]
    errors: Annotated[list[str], operator.add]

Now you need custom exceptions so the pipeline can fail gracefully with clear, agent-specific error messages instead of generic Python tracebacks.

  • Create a new file called core/exceptions.py.
  • Add the exception hierarchy by pasting this code:
class PipelineError(Exception):
    """Base exception for the intelligence pipeline."""
    def __init__(self, message: str, agent: str = "unknown"):
        self.agent = agent
        super().__init__(f"[{agent}] {message}")


class GuardrailError(PipelineError):
    """Raised when input or output fails validation."""
    pass


class ResearchError(PipelineError):
    """Raised when research fails or returns no results."""
    pass


class TokenBudgetExceededError(PipelineError):
    """Raised when an agent exceeds its token budget."""
    pass

What does this code do?

PipelineError is the base exception that automatically tags every error with the agent that raised it. The f"[{agent}] {message}" format makes log messages immediately traceable to a specific node.

The three subclasses (GuardrailError, ResearchError, TokenBudgetExceededError) let you catch specific failure types and respond differently. For example, a GuardrailError might reject the query, while a ResearchError might trigger a retry.

  • Save core/exceptions.py.
  • Verify the exceptions work by running:
python -c "from core.exceptions import GuardrailError; print(GuardrailError('test', agent='guardrails'))"
  • You should see [guardrails] test printed in the terminal.

Seeing a TypeError?

Make sure the __init__ method in PipelineError matches exactly. The agent parameter must have a default value of "unknown".

Help me fix my custom exception classes in core/exceptions.py

✔️ Awesome, I've got everything!

Great. Make sure you've saved the file before moving on.

ⓧ I'd like to double check the full code

class PipelineError(Exception):
    """Base exception for the intelligence pipeline."""
    def __init__(self, message: str, agent: str = "unknown"):
        self.agent = agent
        super().__init__(f"[{agent}] {message}")


class GuardrailError(PipelineError):
    """Raised when input or output fails validation."""
    pass


class ResearchError(PipelineError):
    """Raised when research fails or returns no results."""
    pass


class TokenBudgetExceededError(PipelineError):
    """Raised when an agent exceeds its token budget."""
    pass
Build the structured logging and token cost tracker

Production AI systems need observability. You need to know which agent ran, how long it took, and how many tokens it consumed.

This module gives every agent a structured JSON logger and a shared cost tracker that accumulates OpenAI token usage per node.

  • Create a new file called core/logging_config.py.
  • Add the JSON formatter and logger factory by pasting this code:
import logging
import json
import time
from typing import Any


class JSONFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        log_entry = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "agent": getattr(record, "agent", "system"),
            "action": getattr(record, "action", ""),
            "tokens": getattr(record, "tokens", 0),
            "latency_ms": getattr(record, "latency_ms", 0),
            "message": record.getMessage(),
        }
        return json.dumps(log_entry)


def get_logger(name: str) -> logging.Logger:
    logger = logging.getLogger(name)
    if not logger.handlers:
        handler = logging.StreamHandler()
        handler.setFormatter(JSONFormatter())
        logger.addHandler(handler)
        logger.setLevel(logging.INFO)
    return logger

What does this code do?

JSONFormatter overrides Python's default log formatting to output structured JSON. Each log entry includes the timestamp, agent name, action performed, token count, and latency in milliseconds.

get_logger() is a factory that creates named loggers with the JSON formatter already attached. The if not logger.handlers check prevents duplicate handlers if a logger is requested multiple times.

  • You won't see output yet. The next chunk builds the TokenTracker class that works alongside this logger.

Now add the token cost tracker that accumulates usage from each OpenAI API call and calculates cost using GPT-4o pricing.

  • In core/logging_config.py, add the TokenTracker class below the get_logger function:
class TokenTracker:
    def __init__(self):
        self.usage: dict[str, dict[str, int]] = {}

    def record(self, agent: str, input_tokens: int, output_tokens: int) -> None:
        if agent not in self.usage:
            self.usage[agent] = {"input_tokens": 0, "output_tokens": 0}
        self.usage[agent]["input_tokens"] += input_tokens
        self.usage[agent]["output_tokens"] += output_tokens

    def get_cost(self, agent: str) -> float:
        from config import INPUT_PRICE_PER_1M, OUTPUT_PRICE_PER_1M
        if agent not in self.usage:
            return 0.0
        data = self.usage[agent]
        input_cost = (data["input_tokens"] / 1_000_000) * INPUT_PRICE_PER_1M
        output_cost = (data["output_tokens"] / 1_000_000) * OUTPUT_PRICE_PER_1M
        return input_cost + output_cost

    def get_total_cost(self) -> float:
        return sum(self.get_cost(agent) for agent in self.usage)

    def summary(self) -> str:
        lines = ["--- Token Usage Summary ---"]
        for agent, data in self.usage.items():
            cost = self.get_cost(agent)
            lines.append(
                f"  {agent}: {data['input_tokens']} in / {data['output_tokens']} out (${cost:.4f})"
            )
        lines.append(f"  TOTAL COST: ${self.get_total_cost():.4f}")
        return "\n".join(lines)

What does this code do?

TokenTracker accumulates input and output token counts per agent. Each time an agent makes an API call, it calls tracker.record() with the usage data from the response.

get_cost() calculates the dollar cost using GPT-4o pricing: $2.50 per 1M input tokens and $10.00 per 1M output tokens. The summary() method prints a formatted table at the end of a pipeline run.

  • Save core/logging_config.py.
  • Test the token tracker by running:
python -c "from core.logging_config import TokenTracker; t = TokenTracker(); t.record('planner', 500, 200); print(t.summary())"
  • You should see a token usage summary with the planner's input/output counts and a cost figure.

Seeing an ImportError from config?

The get_cost() method imports from config.py at runtime. Make sure your config.py file exists in the project root (not inside core/) and that you're running the test from the bi-pipeline directory.

Help me debug my TokenTracker import error

✔️ Awesome, I've got everything!

Great. Make sure you've saved the file before moving on.

ⓧ I'd like to double check the full code

import logging
import json
import time
from typing import Any


class JSONFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        log_entry = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "agent": getattr(record, "agent", "system"),
            "action": getattr(record, "action", ""),
            "tokens": getattr(record, "tokens", 0),
            "latency_ms": getattr(record, "latency_ms", 0),
            "message": record.getMessage(),
        }
        return json.dumps(log_entry)


def get_logger(name: str) -> logging.Logger:
    logger = logging.getLogger(name)
    if not logger.handlers:
        handler = logging.StreamHandler()
        handler.setFormatter(JSONFormatter())
        logger.addHandler(handler)
        logger.setLevel(logging.INFO)
    return logger


class TokenTracker:
    def __init__(self):
        self.usage: dict[str, dict[str, int]] = {}

    def record(self, agent: str, input_tokens: int, output_tokens: int) -> None:
        if agent not in self.usage:
            self.usage[agent] = {"input_tokens": 0, "output_tokens": 0}
        self.usage[agent]["input_tokens"] += input_tokens
        self.usage[agent]["output_tokens"] += output_tokens

    def get_cost(self, agent: str) -> float:
        from config import INPUT_PRICE_PER_1M, OUTPUT_PRICE_PER_1M
        if agent not in self.usage:
            return 0.0
        data = self.usage[agent]
        input_cost = (data["input_tokens"] / 1_000_000) * INPUT_PRICE_PER_1M
        output_cost = (data["output_tokens"] / 1_000_000) * OUTPUT_PRICE_PER_1M
        return input_cost + output_cost

    def get_total_cost(self) -> float:
        return sum(self.get_cost(agent) for agent in self.usage)

    def summary(self) -> str:
        lines = ["--- Token Usage Summary ---"]
        for agent, data in self.usage.items():
            cost = self.get_cost(agent)
            lines.append(
                f"  {agent}: {data['input_tokens']} in / {data['output_tokens']} out (${cost:.4f})"
            )
        lines.append(f"  TOTAL COST: ${self.get_total_cost():.4f}")
        return "\n".join(lines)
Build the input and output guardrails

Guardrails protect your pipeline from two directions: bad input going in and bad output coming out. Without them, a too-short query wastes API credits and a report missing key sections looks broken.

  • Create a new file called core/guardrails.py.
  • Add the input validation function by pasting this code:
from core.exceptions import GuardrailError

BLOCKED_TOPICS = ["illegal", "hack", "exploit", "weapon", "drug"]


def validate_input(query: str) -> str:
    query = query.strip()
    if len(query) < 10:
        raise GuardrailError("Query too short. Minimum 10 characters.", agent="guardrails")
    if len(query) > 500:
        raise GuardrailError("Query too long. Maximum 500 characters.", agent="guardrails")
    for topic in BLOCKED_TOPICS:
        if topic.lower() in query.lower():
            raise GuardrailError(
                f"Query contains blocked topic: '{topic}'.", agent="guardrails"
            )
    return query

What does this code do?

validate_input() enforces three rules before any API call happens: the query must be at least 10 characters, no longer than 500 characters, and must not contain any blocked topics.

If any check fails, it raises a GuardrailError with a clear message. This prevents wasting tokens on queries that would produce useless results.

  • You won't see output yet. The next chunk adds the output validation that checks the final report.

Now add the output guardrail that validates the final report before it's shown to the user.

  • In core/guardrails.py, add the validate_output function below validate_input:
def validate_output(report: str, query: str) -> str:
    if len(report) < 200:
        raise GuardrailError(
            "Report too short. Minimum 200 characters.", agent="guardrails"
        )
    required_sections = ["Executive Summary", "Key Findings"]
    for section in required_sections:
        if section.lower() not in report.lower():
            raise GuardrailError(
                f"Report missing required section: '{section}'.", agent="guardrails"
            )
    return report

What does this code do?

validate_output() checks the final report before it reaches the user. It enforces a minimum length (200 characters) and verifies that required sections like "Executive Summary" and "Key Findings" are present.

This catches cases where the LLM produces an incomplete response. Without this, a half-finished report could slip through as the final deliverable.

  • Save core/guardrails.py.
  • Test the guardrails by running:
python -c "from core.guardrails import validate_input; validate_input('short')"
  • You should see a GuardrailError traceback ending with [guardrails] Query too short. Minimum 10 characters. in the terminal. This confirms your guardrails are catching invalid input.

Not seeing the error message?

If Python runs without any output, make sure you typed 'short' (5 characters) as the test input. A longer string like 'analyze market trends' would pass validation and return silently.

If you see an import error, check that core/exceptions.py exists and has the GuardrailError class defined.

Help me debug my guardrails module

✔️ Awesome, I've got everything!

Great. Make sure you've saved the file before moving on.

ⓧ I'd like to double check the full code

from core.exceptions import GuardrailError

BLOCKED_TOPICS = ["illegal", "hack", "exploit", "weapon", "drug"]


def validate_input(query: str) -> str:
    query = query.strip()
    if len(query) < 10:
        raise GuardrailError("Query too short. Minimum 10 characters.", agent="guardrails")
    if len(query) > 500:
        raise GuardrailError("Query too long. Maximum 500 characters.", agent="guardrails")
    for topic in BLOCKED_TOPICS:
        if topic.lower() in query.lower():
            raise GuardrailError(
                f"Query contains blocked topic: '{topic}'.", agent="guardrails"
            )
    return query


def validate_output(report: str, query: str) -> str:
    if len(report) < 200:
        raise GuardrailError(
            "Report too short. Minimum 200 characters.", agent="guardrails"
        )
    required_sections = ["Executive Summary", "Key Findings"]
    for section in required_sections:
        if section.lower() not in report.lower():
            raise GuardrailError(
                f"Report missing required section: '{section}'.", agent="guardrails"
            )
    return report

Your production foundation is solid. You have typed state, custom exceptions, structured logging, cost tracking, and guardrails all wired up. Next up, you'll build the four specialized agent nodes that use this infrastructure to research, analyze, and report.

Build the Agent Nodes

Your production foundation is solid. You have validated state, custom exceptions, structured logging, and guardrails all ready to go.

Now it's time to build the agents that actually do the work. Each agent is a specialized function that receives the pipeline state, calls an LLM or tool, tracks its token usage, and returns state updates.

In this step, you'll create four agent nodes: a Planner that breaks questions into sub-tasks, a Researcher that searches the web and compresses findings, an Analyst that scores research quality, and a Report Writer that produces the final brief.

In this step, get ready to:
  • Build the Planner Agent to decompose business questions into focused research sub-tasks.
  • Build the Researcher Agent with Tavily web search and context compression.
  • Build the Analyst and Report Writer Agents for quality scoring and report generation.
Build the Planner Agent

The first agent in the pipeline takes the user's business question and decomposes it into 3-5 focused research sub-tasks. This is a token optimization strategy.

Instead of sending the full question directly to a search tool, the Planner produces concise, actionable sub-queries that keep downstream token usage low.

  • Create a new file called agents/planner.py inside your bi-pipeline project folder by pasting the following code:
import time
from langchain_openai import ChatOpenAI
from core.logging_config import get_logger, TokenTracker
from core.state import PipelineState
from config import MODEL_NAME

logger = get_logger("planner")


def planner_node(state: PipelineState, tracker: TokenTracker) -> dict:
    start = time.time()
    llm = ChatOpenAI(model=MODEL_NAME, temperature=0)

    prompt = f"""You are a research planner for business intelligence.
Given this business question, decompose it into 3-5 focused research sub-tasks.
Each sub-task must be under 50 words and directly actionable as a web search query.
Output ONLY a numbered list. No preamble.

Business Question: {state['query']}"""

    response = llm.invoke(prompt)
    latency = int((time.time() - start) * 1000)

    input_tokens = response.response_metadata.get("token_usage", {}).get("prompt_tokens", 0)
    output_tokens = response.response_metadata.get("token_usage", {}).get("completion_tokens", 0)
    tracker.record("planner", input_tokens, output_tokens)

    plan = [line.strip() for line in response.content.strip().split("\n") if line.strip()]

    logger.info(
        "Research plan generated",
        extra={"agent": "planner", "action": "decompose", "tokens": input_tokens + output_tokens, "latency_ms": latency},
    )

    return {"research_plan": plan}

What does this code do?

The planner_node function takes the pipeline state and a TokenTracker instance. It sends the business question to GPT-4o with a prompt that constrains each sub-task to under 50 words.

After the LLM responds, it extracts token counts from response.response_metadata and records them with the tracker. The result is parsed into a list of strings and returned as research_plan.

  • Save agents/planner.py.
  • Verify the module loads correctly by running:
python -c "from agents.planner import planner_node; print('Planner agent loaded OK')"

You should see Planner agent loaded OK printed in your terminal.

Seeing a ModuleNotFoundError?

Make sure you're running the command from the bi-pipeline directory (the one containing the agents/ and core/ folders). Also confirm your virtual environment is active.

If the error mentions langchain_openai, double check that you installed langchain-openai in your virtual environment during Step 1.

help me fix a ModuleNotFoundError when importing my planner agent

✔️ Awesome, I've got everything!

Great. Make sure you've saved agents/planner.py before moving on.

ⓧ I'd like to double check the full code

import time
from langchain_openai import ChatOpenAI
from core.logging_config import get_logger, TokenTracker
from core.state import PipelineState
from config import MODEL_NAME

logger = get_logger("planner")


def planner_node(state: PipelineState, tracker: TokenTracker) -> dict:
    start = time.time()
    llm = ChatOpenAI(model=MODEL_NAME, temperature=0)

    prompt = f"""You are a research planner for business intelligence.
Given this business question, decompose it into 3-5 focused research sub-tasks.
Each sub-task must be under 50 words and directly actionable as a web search query.
Output ONLY a numbered list. No preamble.

Business Question: {state['query']}"""

    response = llm.invoke(prompt)
    latency = int((time.time() - start) * 1000)

    input_tokens = response.response_metadata.get("token_usage", {}).get("prompt_tokens", 0)
    output_tokens = response.response_metadata.get("token_usage", {}).get("completion_tokens", 0)
    tracker.record("planner", input_tokens, output_tokens)

    plan = [line.strip() for line in response.content.strip().split("\n") if line.strip()]

    logger.info(
        "Research plan generated",
        extra={"agent": "planner", "action": "decompose", "tokens": input_tokens + output_tokens, "latency_ms": latency},
    )

    return {"research_plan": plan}
Build the Researcher Agent with web search and compression

The Researcher is the most complex agent. It takes the sub-tasks from the Planner, searches the web using Tavily, and collects raw results.

Here's the token optimization pattern: raw search results are verbose. Before passing them to the Analyst, the Researcher compresses the findings into a concise briefing under 800 words. This reduces downstream token costs significantly.

  • Create a new file called agents/researcher.py and paste the following code:
import time
from langchain_openai import ChatOpenAI
from langchain_tavily import TavilySearch
from core.logging_config import get_logger, TokenTracker
from core.state import PipelineState
from config import MODEL_NAME

logger = get_logger("researcher")


def researcher_node(state: PipelineState, tracker: TokenTracker) -> dict:
    start = time.time()
    search_tool = TavilySearch(max_results=3, topic="general")
    llm = ChatOpenAI(model=MODEL_NAME, temperature=0)

    all_results = []
    for task in state["research_plan"]:
        try:
            results = search_tool.invoke({"query": task})
            all_results.append(f"## {task}\n{results}")
        except Exception as e:
            all_results.append(f"## {task}\nSearch failed: {str(e)}")

    raw_findings = "\n\n".join(all_results)

    compress_prompt = f"""Summarize the following research findings into a concise briefing.
Keep only the most relevant facts, data points, and sources.
Target: under 800 words. Preserve source URLs where available.

Research Findings:
{raw_findings[:8000]}"""

    compress_response = llm.invoke(compress_prompt)
    latency = int((time.time() - start) * 1000)

    input_tokens = compress_response.response_metadata.get("token_usage", {}).get("prompt_tokens", 0)
    output_tokens = compress_response.response_metadata.get("token_usage", {}).get("completion_tokens", 0)
    tracker.record("researcher", input_tokens, output_tokens)

    logger.info(
        "Research complete and compressed",
        extra={"agent": "researcher", "action": "research_and_compress", "tokens": input_tokens + output_tokens, "latency_ms": latency},
    )

    return {
        "raw_findings": raw_findings[:3000],
        "compressed_findings": compress_response.content,
    }

What does this code do?

The researcher_node function does two things. First, it loops through each sub-task from the Planner and runs a TavilySearch query with max_results=3 to collect web results. Failed searches are caught gracefully so one bad query doesn't crash the pipeline.

Second, it sends the raw findings to GPT-4o with a compression prompt. The LLM distills thousands of tokens of raw search results into a concise briefing under 800 words. This is the token optimization layer. The Analyst only receives the compressed version, not the raw results.

Why compress before passing to the next agent?

Raw search results from Tavily can easily exceed 5,000 tokens per sub-task. With 3-5 sub-tasks, the Analyst would receive 15,000-25,000 tokens of input. Compression reduces this to under 800 words (roughly 1,000 tokens).

This is an architectural decision, not just prompt engineering. Each agent is responsible for handing off only what the next agent needs.

  • Save agents/researcher.py.
  • Verify the module loads correctly by running:
python -c "from agents.researcher import researcher_node; print('Researcher agent loaded OK')"

You should see Researcher agent loaded OK printed in your terminal.

Seeing an import error for langchain_tavily?

Confirm that langchain-tavily is installed in your virtual environment. The package name uses a hyphen (langchain-tavily) but the import uses an underscore (langchain_tavily).

help me fix an import error for langchain_tavily in my researcher agent

✔️ Awesome, I've got everything!

Great. Make sure you've saved agents/researcher.py before moving on.

ⓧ I'd like to double check the full code

import time
from langchain_openai import ChatOpenAI
from langchain_tavily import TavilySearch
from core.logging_config import get_logger, TokenTracker
from core.state import PipelineState
from config import MODEL_NAME

logger = get_logger("researcher")


def researcher_node(state: PipelineState, tracker: TokenTracker) -> dict:
    start = time.time()
    search_tool = TavilySearch(max_results=3, topic="general")
    llm = ChatOpenAI(model=MODEL_NAME, temperature=0)

    all_results = []
    for task in state["research_plan"]:
        try:
            results = search_tool.invoke({"query": task})
            all_results.append(f"## {task}\n{results}")
        except Exception as e:
            all_results.append(f"## {task}\nSearch failed: {str(e)}")

    raw_findings = "\n\n".join(all_results)

    compress_prompt = f"""Summarize the following research findings into a concise briefing.
Keep only the most relevant facts, data points, and sources.
Target: under 800 words. Preserve source URLs where available.

Research Findings:
{raw_findings[:8000]}"""

    compress_response = llm.invoke(compress_prompt)
    latency = int((time.time() - start) * 1000)

    input_tokens = compress_response.response_metadata.get("token_usage", {}).get("prompt_tokens", 0)
    output_tokens = compress_response.response_metadata.get("token_usage", {}).get("completion_tokens", 0)
    tracker.record("researcher", input_tokens, output_tokens)

    logger.info(
        "Research complete and compressed",
        extra={"agent": "researcher", "action": "research_and_compress", "tokens": input_tokens + output_tokens, "latency_ms": latency},
    )

    return {
        "raw_findings": raw_findings[:3000],
        "compressed_findings": compress_response.content,
    }
Build the Analyst and Report Writer Agents

The last two agents work with the compressed context produced by the Researcher. The Analyst interprets findings and assigns a quality score. The Report Writer synthesizes everything into a structured intelligence brief.

Both agents receive only the compressed findings, not raw search results. This is the payoff of the token optimization you built into the Researcher.

  • Create a new file called agents/analyst.py and paste the following code:
import time
from langchain_openai import ChatOpenAI
from core.logging_config import get_logger, TokenTracker
from core.state import PipelineState
from config import MODEL_NAME

logger = get_logger("analyst")


def analyst_node(state: PipelineState, tracker: TokenTracker) -> dict:
    start = time.time()
    llm = ChatOpenAI(model=MODEL_NAME, temperature=0)

    prompt = f"""You are a business analyst. Analyze these compressed research findings.

1. Identify key patterns, trends, and competitive dynamics.
2. Flag any risks or gaps in the data.
3. Rate the overall research quality from 1-10 based on:
   - Source diversity (multiple sources?)
   - Relevance to the original question
   - Recency of information
   - Depth of coverage

Original Question: {state['query']}

Compressed Findings:
{state['compressed_findings']}

Output format:
ANALYSIS: [your analysis]
QUALITY_SCORE: [1-10]"""

    response = llm.invoke(prompt)
    latency = int((time.time() - start) * 1000)

    input_tokens = response.response_metadata.get("token_usage", {}).get("prompt_tokens", 0)
    output_tokens = response.response_metadata.get("token_usage", {}).get("completion_tokens", 0)
    tracker.record("analyst", input_tokens, output_tokens)

    content = response.content
    quality_score = 7
    if "QUALITY_SCORE:" in content:
        try:
            score_str = content.split("QUALITY_SCORE:")[1].strip().split("\n")[0].strip()
            quality_score = int(score_str)
        except (ValueError, IndexError):
            quality_score = 7

    analysis = content.split("QUALITY_SCORE:")[0].replace("ANALYSIS:", "").strip() if "QUALITY_SCORE:" in content else content

    logger.info(
        f"Analysis complete. Quality score: {quality_score}",
        extra={"agent": "analyst", "action": "analyze", "tokens": input_tokens + output_tokens, "latency_ms": latency},
    )

    return {"analysis": analysis, "quality_score": quality_score}

What does this code do?

The analyst_node asks GPT-4o to analyze compressed research findings and produce two outputs: a written analysis and a numeric quality score from 1-10.

The parsing logic splits the response at QUALITY_SCORE: to extract the score as an integer. If parsing fails for any reason, it defaults to 7. This quality score drives the conditional routing you'll build in the next step.

  • Save agents/analyst.py.
  • Verify it loads by running:
python -c "from agents.analyst import analyst_node; print('Analyst agent loaded OK')"

You should see Analyst agent loaded OK in the terminal output.

Seeing a SyntaxError in the f-string?

The triple-quoted f-string in the prompt uses curly braces for variable interpolation. Make sure you haven't accidentally added extra braces or removed the f prefix from the string. Python 3.10+ supports multi-line f-strings but the braces must match exactly.

help me fix a SyntaxError in my analyst agent f-string prompt

✔️ Awesome, I've got everything!

Great. Make sure you've saved agents/analyst.py before moving on to the Report Writer.

ⓧ I'd like to double check the full code

import time
from langchain_openai import ChatOpenAI
from core.logging_config import get_logger, TokenTracker
from core.state import PipelineState
from config import MODEL_NAME

logger = get_logger("analyst")


def analyst_node(state: PipelineState, tracker: TokenTracker) -> dict:
    start = time.time()
    llm = ChatOpenAI(model=MODEL_NAME, temperature=0)

    prompt = f"""You are a business analyst. Analyze these compressed research findings.

1. Identify key patterns, trends, and competitive dynamics.
2. Flag any risks or gaps in the data.
3. Rate the overall research quality from 1-10 based on:
   - Source diversity (multiple sources?)
   - Relevance to the original question
   - Recency of information
   - Depth of coverage

Original Question: {state['query']}

Compressed Findings:
{state['compressed_findings']}

Output format:
ANALYSIS: [your analysis]
QUALITY_SCORE: [1-10]"""

    response = llm.invoke(prompt)
    latency = int((time.time() - start) * 1000)

    input_tokens = response.response_metadata.get("token_usage", {}).get("prompt_tokens", 0)
    output_tokens = response.response_metadata.get("token_usage", {}).get("completion_tokens", 0)
    tracker.record("analyst", input_tokens, output_tokens)

    content = response.content
    quality_score = 7
    if "QUALITY_SCORE:" in content:
        try:
            score_str = content.split("QUALITY_SCORE:")[1].strip().split("\n")[0].strip()
            quality_score = int(score_str)
        except (ValueError, IndexError):
            quality_score = 7

    analysis = content.split("QUALITY_SCORE:")[0].replace("ANALYSIS:", "").strip() if "QUALITY_SCORE:" in content else content

    logger.info(
        f"Analysis complete. Quality score: {quality_score}",
        extra={"agent": "analyst", "action": "analyze", "tokens": input_tokens + output_tokens, "latency_ms": latency},
    )

    return {"analysis": analysis, "quality_score": quality_score}

Now for the final agent. The Report Writer takes the analysis and compressed research, then produces a structured markdown report.

  • Create a new file called agents/report_writer.py and paste the following code:
import time
from langchain_openai import ChatOpenAI
from core.logging_config import get_logger, TokenTracker
from core.state import PipelineState
from config import MODEL_NAME

logger = get_logger("report_writer")


def report_writer_node(state: PipelineState, tracker: TokenTracker) -> dict:
    start = time.time()
    llm = ChatOpenAI(model=MODEL_NAME, temperature=0.2)

    prompt = f"""You are a senior business intelligence report writer.
Write a structured intelligence brief based on the analysis below.

Required sections:
1. Executive Summary (2-3 sentences)
2. Key Findings (3-5 bullet points with supporting evidence)
3. Risks and Uncertainties
4. Recommendations (3 actionable next steps)

Original Question: {state['query']}

Analysis:
{state['analysis']}

Compressed Research:
{state['compressed_findings'][:2000]}

Write in professional, concise language. Cite sources where available."""

    response = llm.invoke(prompt)
    latency = int((time.time() - start) * 1000)

    input_tokens = response.response_metadata.get("token_usage", {}).get("prompt_tokens", 0)
    output_tokens = response.response_metadata.get("token_usage", {}).get("completion_tokens", 0)
    tracker.record("report_writer", input_tokens, output_tokens)

    logger.info(
        "Report generated",
        extra={"agent": "report_writer", "action": "write_report", "tokens": input_tokens + output_tokens, "latency_ms": latency},
    )

    return {"report": response.content}

What does this code do?

The report_writer_node uses a slightly higher temperature (0.2) than the other agents. This allows for more natural, varied writing while still keeping the output structured.

Notice it only receives state['compressed_findings'][:2000] instead of raw search results. Combined with the analysis, this gives the Report Writer everything it needs with minimal token overhead.

  • Save agents/report_writer.py.
  • Verify all four agents load correctly by running:
python -c "from agents.planner import planner_node; from agents.researcher import researcher_node; from agents.analyst import analyst_node; from agents.report_writer import report_writer_node; print('All 4 agents loaded OK')"

You should see All 4 agents loaded OK printed in your terminal. This confirms all imports resolve and there are no syntax errors.

One of the agents failing to import?

Check which agent is causing the failure. Python will show the exact file and line number in the traceback. Common causes are a missing comma in a dictionary, an unclosed parenthesis, or a mismatched quote inside an f-string.

help me fix import errors across my agent files

✔️ Awesome, I've got everything!

Great. Make sure you've saved agents/report_writer.py before moving on.

ⓧ I'd like to double check the full code

import time
from langchain_openai import ChatOpenAI
from core.logging_config import get_logger, TokenTracker
from core.state import PipelineState
from config import MODEL_NAME

logger = get_logger("report_writer")


def report_writer_node(state: PipelineState, tracker: TokenTracker) -> dict:
    start = time.time()
    llm = ChatOpenAI(model=MODEL_NAME, temperature=0.2)

    prompt = f"""You are a senior business intelligence report writer.
Write a structured intelligence brief based on the analysis below.

Required sections:
1. Executive Summary (2-3 sentences)
2. Key Findings (3-5 bullet points with supporting evidence)
3. Risks and Uncertainties
4. Recommendations (3 actionable next steps)

Original Question: {state['query']}

Analysis:
{state['analysis']}

Compressed Research:
{state['compressed_findings'][:2000]}

Write in professional, concise language. Cite sources where available."""

    response = llm.invoke(prompt)
    latency = int((time.time() - start) * 1000)

    input_tokens = response.response_metadata.get("token_usage", {}).get("prompt_tokens", 0)
    output_tokens = response.response_metadata.get("token_usage", {}).get("completion_tokens", 0)
    tracker.record("report_writer", input_tokens, output_tokens)

    logger.info(
        "Report generated",
        extra={"agent": "report_writer", "action": "write_report", "tokens": input_tokens + output_tokens, "latency_ms": latency},
    )

    return {"report": response.content}

All four agents are built and ready to be orchestrated. Next up, you'll wire them together into a LangGraph StateGraph with conditional routing and human-in-the-loop approval.

Orchestrate the Graph

Your 4 specialized agents are built and ready to work. Each one handles a distinct piece of the intelligence pipeline: planning research tasks, searching the web, analyzing findings, and writing reports.

Now you need an orchestrator that wires them into a directed graph, controls the flow between nodes with conditional routing, and lets a human approve the output before the Report Writer runs. This is where LangGraph's StateGraph shines.

In this step, get ready to:
  • Wire the StateGraph with agent nodes and a human-in-the-loop approval gate.
  • Add conditional routing that loops low-quality research back for refinement.
  • Build the main execution loop with interrupt handling and cost tracking.
Wire the StateGraph with all nodes and the human-in-the-loop approval node

The orchestrator file pulls together every module you have built so far. It imports the agent node functions, wraps them with a shared TokenTracker instance, and defines the graph structure that controls execution order.

  • Open main.py in Cursor (it should be empty) and paste the following imports, globals, and wrapper functions:
from typing import Literal
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt
from langgraph.checkpoint.memory import InMemorySaver

import config
from core.state import PipelineState
from core.guardrails import validate_input, validate_output
from core.logging_config import get_logger, TokenTracker
from core.exceptions import PipelineError, GuardrailError
from agents.planner import planner_node
from agents.researcher import researcher_node
from agents.analyst import analyst_node
from agents.report_writer import report_writer_node

logger = get_logger("orchestrator")
tracker = TokenTracker()


def planner(state: PipelineState) -> dict:
    return planner_node(state, tracker)


def researcher(state: PipelineState) -> dict:
    result = researcher_node(state, tracker)
    loops = state.get("research_loops", 0) + 1
    result["research_loops"] = loops
    return result


def analyst(state: PipelineState) -> dict:
    return analyst_node(state, tracker)


def report_writer(state: PipelineState) -> dict:
    return report_writer_node(state, tracker)

What does this code do?

The imports pull in LangGraph's graph primitives (StateGraph, START, END), the interrupt/resume mechanism (interrupt, Command), and the in-memory checkpointer needed for persistence across interrupts.

Each wrapper function is a thin adapter. LangGraph nodes receive only state, but your agent functions also need the shared tracker instance. The wrappers bridge that gap. The researcher wrapper also increments research_loops so the graph knows how many times it has looped back.

Next, add the human-in-the-loop approval node. This is the function that pauses the entire graph and waits for a human decision before continuing.

  • Directly below the report_writer wrapper, add the human_approval function:
def human_approval(state: PipelineState) -> dict:
    decision = interrupt({
        "question": "Do you approve proceeding to report generation?",
        "quality_score": state["quality_score"],
        "analysis_preview": state["analysis"][:500],
    })
    return {"approval_status": "approved" if decision else "rejected"}

How does interrupt() work?

When the graph reaches this node, interrupt() pauses execution and saves the entire graph state to the checkpointer. The dictionary you pass to interrupt() is surfaced to the caller as metadata describing what the human needs to decide.

Later, when you resume the graph with Command(resume=...), the value you pass becomes the return value of interrupt() inside this function. If the human sends True, decision is truthy and approval_status becomes "approved".

interrupt() must come first in the node

Any code before interrupt() will re-run when the graph resumes. Place the interrupt() call at the very top of the function to avoid unintended side effects.

Add conditional routing after the Analyst node

Conditional routing is what makes this pipeline intelligent. Instead of always running every agent in sequence, the graph inspects the quality score and decides whether to loop back for more research or proceed to human approval.

  • Below the human_approval function, add the two routing functions:
def route_after_analyst(state: PipelineState) -> Literal["researcher", "human_approval"]:
    if state["quality_score"] < config.QUALITY_THRESHOLD and state.get("research_loops", 0) < config.MAX_RESEARCH_LOOPS:
        logger.info("Quality below threshold, re-routing to researcher")
        return "researcher"
    return "human_approval"


def route_after_approval(state: PipelineState) -> Literal["report_writer", "__end__"]:
    if state.get("approval_status") == "approved":
        return "report_writer"
    return "__end__"

What does this code do?

  • route_after_analyst checks two conditions: is the quality score below 6, AND has the graph looped fewer than 2 times? If both are true, it sends control back to the Researcher for another pass. Otherwise, it moves forward to human approval.
  • route_after_approval checks whether the human said yes. If approved, it routes to the Report Writer. If rejected, it routes to "__end__" which terminates the graph without generating a report.
  • The MAX_RESEARCH_LOOPS = 2 cap prevents infinite cycles. Without it, a consistently low-quality research result would loop forever.

Now assemble everything into a compiled graph. The build_graph() function registers nodes, wires edges (including conditional ones), and compiles with a checkpointer.

  • Below the routing functions, add the build_graph() function:
def build_graph():
    builder = StateGraph(PipelineState)

    builder.add_node("planner", planner)
    builder.add_node("researcher", researcher)
    builder.add_node("analyst", analyst)
    builder.add_node("human_approval", human_approval)
    builder.add_node("report_writer", report_writer)

    builder.add_edge(START, "planner")
    builder.add_edge("planner", "researcher")
    builder.add_edge("researcher", "analyst")
    builder.add_conditional_edges("analyst", route_after_analyst)
    builder.add_conditional_edges("human_approval", route_after_approval)
    builder.add_edge("report_writer", END)

    return builder.compile(checkpointer=InMemorySaver())

What does this code do?

This wires the full execution flow. Fixed edges (add_edge) connect nodes that always run in sequence: START to planner, planner to researcher, researcher to analyst, and report_writer to END.

Conditional edges (add_conditional_edges) branch based on what the routing function returns. After the analyst, the graph either loops back to the researcher or moves to human approval. After human approval, it either writes the report or ends.

The InMemorySaver checkpointer is required for interrupt() to work. It persists graph state while the execution is paused waiting for human input.

Why InMemorySaver instead of a database?

For a CLI tool that runs in a single session, in-memory persistence is sufficient. In production, you would swap this for a database-backed checkpointer (like PostgreSQL) so that interrupted runs survive process restarts.

Build the main execution loop with interrupt handling

The main() function is the entry point that ties everything together. It takes a question from the CLI, validates it against the input guardrails, streams the graph, handles the human approval interrupt, validates the output, and prints the final report with a cost breakdown.

  • Below build_graph(), add the main() function and entry point:
def main():
    query = input("Enter your business intelligence question: ")

    try:
        query = validate_input(query)
    except GuardrailError as e:
        print(f"Input validation failed: {e}")
        return

    graph = build_graph()
    thread_config = {"configurable": {"thread_id": "intel-run-1"}}

    initial_state = {
        "query": query,
        "research_plan": [],
        "raw_findings": "",
        "compressed_findings": "",
        "analysis": "",
        "quality_score": 0,
        "report": "",
        "approval_status": "",
        "research_loops": 0,
        "token_usage": {},
        "errors": [],
    }

    print("\n--- Starting Intelligence Pipeline ---\n")

    result = None
    for event in graph.stream(initial_state, thread_config, stream_mode="updates"):
        for node_name, node_output in event.items():
            print(f"[{node_name}] completed")

    snapshot = graph.get_state(thread_config)
    if snapshot.next and "human_approval" in str(snapshot.next):
        interrupts = snapshot.tasks
        if interrupts:
            print("\n--- HUMAN APPROVAL REQUIRED ---")
            print(f"Quality Score: {snapshot.values.get('quality_score', 'N/A')}")
            print(f"Analysis Preview: {snapshot.values.get('analysis', '')[:300]}")
            approval = input("\nApprove report generation? (yes/no): ").strip().lower()

            for event in graph.stream(Command(resume=(approval == "yes")), thread_config, stream_mode="updates"):
                for node_name, node_output in event.items():
                    print(f"[{node_name}] completed")

    final_state = graph.get_state(thread_config)
    report = final_state.values.get("report", "")

    if report:
        try:
            validate_output(report, query)
            print("\n--- INTELLIGENCE REPORT ---\n")
            print(report)
        except GuardrailError as e:
            print(f"Output validation failed: {e}")
    else:
        print("\nPipeline ended without generating a report (rejected or error).")

    print(f"\n{tracker.summary()}")


if __name__ == "__main__":
    main()

What does this code do?

  • validate_input(query) runs the guardrails you built earlier. If the query is too short, too long, or contains a blocked topic, the pipeline exits early with a clear error message.
  • graph.stream() with stream_mode="updates" runs the graph node by node, yielding each node's output as it completes. This gives you real-time progress feedback in the terminal.
  • When the graph hits interrupt() in the human_approval node, streaming pauses. The code then checks snapshot.next to confirm the graph is waiting for human input, displays the quality score and analysis preview, and asks for approval.
  • Command(resume=(approval == "yes")) resumes the graph. The boolean value becomes the return value of interrupt() inside human_approval().
  • validate_output() checks the final report for minimum length and required sections. The tracker summary prints per-agent token usage and total cost.
  • Save main.py.

Seeing an import error?

  • Make sure your virtual environment is activated. The langgraph package must be installed in the active environment.
  • Check that your agents/ and core/ directories each have an __init__.py file (even if empty). Python needs these to treat the folders as packages.
  • Verify that function names match exactly. The imports expect planner_node, researcher_node, analyst_node, and report_writer_node in their respective files.

Help me debug an import error in my LangGraph pipeline

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py before moving on.

ⓧ I'd like to double check the full code

from typing import Literal
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt
from langgraph.checkpoint.memory import InMemorySaver

import config
from core.state import PipelineState
from core.guardrails import validate_input, validate_output
from core.logging_config import get_logger, TokenTracker
from core.exceptions import PipelineError, GuardrailError
from agents.planner import planner_node
from agents.researcher import researcher_node
from agents.analyst import analyst_node
from agents.report_writer import report_writer_node

logger = get_logger("orchestrator")
tracker = TokenTracker()


def planner(state: PipelineState) -> dict:
    return planner_node(state, tracker)


def researcher(state: PipelineState) -> dict:
    result = researcher_node(state, tracker)
    loops = state.get("research_loops", 0) + 1
    result["research_loops"] = loops
    return result


def analyst(state: PipelineState) -> dict:
    return analyst_node(state, tracker)


def human_approval(state: PipelineState) -> dict:
    decision = interrupt({
        "question": "Do you approve proceeding to report generation?",
        "quality_score": state["quality_score"],
        "analysis_preview": state["analysis"][:500],
    })
    return {"approval_status": "approved" if decision else "rejected"}


def report_writer(state: PipelineState) -> dict:
    return report_writer_node(state, tracker)


def route_after_analyst(state: PipelineState) -> Literal["researcher", "human_approval"]:
    if state["quality_score"] < config.QUALITY_THRESHOLD and state.get("research_loops", 0) < config.MAX_RESEARCH_LOOPS:
        logger.info("Quality below threshold, re-routing to researcher")
        return "researcher"
    return "human_approval"


def route_after_approval(state: PipelineState) -> Literal["report_writer", "__end__"]:
    if state.get("approval_status") == "approved":
        return "report_writer"
    return "__end__"


def build_graph():
    builder = StateGraph(PipelineState)

    builder.add_node("planner", planner)
    builder.add_node("researcher", researcher)
    builder.add_node("analyst", analyst)
    builder.add_node("human_approval", human_approval)
    builder.add_node("report_writer", report_writer)

    builder.add_edge(START, "planner")
    builder.add_edge("planner", "researcher")
    builder.add_edge("researcher", "analyst")
    builder.add_conditional_edges("analyst", route_after_analyst)
    builder.add_conditional_edges("human_approval", route_after_approval)
    builder.add_edge("report_writer", END)

    return builder.compile(checkpointer=InMemorySaver())


def main():
    query = input("Enter your business intelligence question: ")

    try:
        query = validate_input(query)
    except GuardrailError as e:
        print(f"Input validation failed: {e}")
        return

    graph = build_graph()
    thread_config = {"configurable": {"thread_id": "intel-run-1"}}

    initial_state = {
        "query": query,
        "research_plan": [],
        "raw_findings": "",
        "compressed_findings": "",
        "analysis": "",
        "quality_score": 0,
        "report": "",
        "approval_status": "",
        "research_loops": 0,
        "token_usage": {},
        "errors": [],
    }

    print("\n--- Starting Intelligence Pipeline ---\n")

    result = None
    for event in graph.stream(initial_state, thread_config, stream_mode="updates"):
        for node_name, node_output in event.items():
            print(f"[{node_name}] completed")

    snapshot = graph.get_state(thread_config)
    if snapshot.next and "human_approval" in str(snapshot.next):
        interrupts = snapshot.tasks
        if interrupts:
            print("\n--- HUMAN APPROVAL REQUIRED ---")
            print(f"Quality Score: {snapshot.values.get('quality_score', 'N/A')}")
            print(f"Analysis Preview: {snapshot.values.get('analysis', '')[:300]}")
            approval = input("\nApprove report generation? (yes/no): ").strip().lower()

            for event in graph.stream(Command(resume=(approval == "yes")), thread_config, stream_mode="updates"):
                for node_name, node_output in event.items():
                    print(f"[{node_name}] completed")

    final_state = graph.get_state(thread_config)
    report = final_state.values.get("report", "")

    if report:
        try:
            validate_output(report, query)
            print("\n--- INTELLIGENCE REPORT ---\n")
            print(report)
        except GuardrailError as e:
            print(f"Output validation failed: {e}")
    else:
        print("\nPipeline ended without generating a report (rejected or error).")

    print(f"\n{tracker.summary()}")


if __name__ == "__main__":
    main()

Your orchestrator is complete. All 4 agents are wired into a graph with conditional routing and human-in-the-loop approval. Next up, you will run the full pipeline end-to-end and verify everything works in LangSmith.

Run, Trace, and Validate the Pipeline

You have a complete multi-agent pipeline wired up and ready to go. Now it is time to run it end-to-end, verify that every agent traces correctly in LangSmith, and stress-test the guardrails you built.

This step is all about validation. You will confirm that the graph executes in the correct order, that the human-in-the-loop interrupt works, that invalid inputs get caught, and that the token optimization actually reduces context size between agents.

In this step, get ready to:
  • Run the full pipeline with a real business question and approve report generation.
  • Verify traces in LangSmith and test guardrails with invalid inputs.
  • Review structured logs, confirm token optimization, and print a cost breakdown.
Run the full pipeline end-to-end

Your pipeline takes a business question, decomposes it into research tasks, searches the web, analyzes findings, asks for human approval, and generates a structured report. Let's watch it work.

  • Make sure your virtual environment is activated, then run the pipeline by executing this command:
python main.py
  • When prompted, enter this sample business question:
Analyze the competitive landscape of AI coding assistants

The pipeline will take 30-60 seconds as the agents work through their tasks. You will see structured JSON log output for each agent completing its work.

  • Watch the terminal output. You should see each node complete in sequence: [planner], [researcher], [analyst].
  • When the --- HUMAN APPROVAL REQUIRED --- prompt appears, review the quality score and analysis preview displayed.
  • Type yes and press Enter to approve report generation.

What is happening under the hood?

The interrupt() call in human_approval pauses graph execution and saves state via the InMemorySaver checkpointer. When you type yes, main.py resumes the graph with Command(resume=True), which becomes the return value of interrupt() inside the paused node.

This is the human-in-the-loop pattern. The graph literally freezes mid-execution and picks up exactly where it left off after the human makes a decision.

  • After approval, the [report_writer] and [human_approval] nodes complete. You should see --- INTELLIGENCE REPORT --- followed by a structured report with Executive Summary, Key Findings, Risks, and Recommendations sections.
  • At the very end of the output, you should see --- Token Usage Summary --- showing per-agent token counts and a total cost in dollars.

What should the cost look like?

A typical run costs between $0.01 and $0.05 depending on the complexity of the question and search results. The planner uses the fewest tokens. The researcher uses the most because it processes raw search results before compression.

Now verify your trace appeared in LangSmith.

  • Open your browser and go to https://smith.langchain.com.
  • In the LangSmith UI, click Tracing in the sidebar.
  • Select the business-intel-pipeline project from the project list.
  • Click the most recent trace row to open it.
  • Switch to the Details tab to see the full run tree with each agent node, token counts, and latency for every step.

What should I see in the trace?

Your trace shows the complete execution path. You should see nested spans for the planner, researcher, analyst, human_approval, and report_writer nodes. Inside the researcher span, you can see individual Tavily tool calls.

The Details view renders token counts and latency for each run. This is the full observability your production system needs.

Trace not appearing in LangSmith?

Confirm that your .env file has LANGSMITH_TRACING=true and a valid LANGSMITH_API_KEY. Also verify that LANGSMITH_PROJECT=business-intel-pipeline matches the project you are looking at in the UI.

help me debug why my LangSmith traces are not appearing for my LangGraph pipeline

Test the guardrails and conditional routing

Production systems need to handle bad inputs gracefully. Your guardrails module validates queries before they enter the pipeline. Let's confirm they catch invalid inputs and that the conditional routing loop works.

  • Run the pipeline again to test the minimum length guardrail:
python main.py
  • When prompted for a question, enter a query that is too short (under 10 characters):
Hi
  • You should see: Input validation failed: [guardrails] Query too short. Minimum 10 characters. The pipeline exits immediately without calling any agents.
  • Run the pipeline once more to test the topic blocklist:
python main.py
  • Enter a query containing a blocked keyword:
How to exploit vulnerabilities in web applications
  • You should see: Input validation failed: [guardrails] Query contains blocked topic: 'exploit'. The GuardrailError identifies exactly which blocked word triggered the rejection.

What about the conditional re-research loop?

The route_after_analyst function checks if quality_score < 6 and research_loops < 2. If both conditions are true, it routes back to the researcher for another pass. In practice, most queries score 6 or above on the first attempt because Tavily returns high-quality results.

If you want to observe the loop in action, you can temporarily change QUALITY_THRESHOLD in config.py to 9. This forces the analyst to almost always trigger a re-research cycle. Remember to set it back to 6 afterwards.

Getting a different error?

If you see an error about TAVILY_API_KEY or OPENAI_API_KEY, double-check that your .env file has valid keys (not placeholder values) and that your virtual environment is activated.

help me debug API key errors when running my LangGraph multi-agent pipeline

Review logs, verify token optimization, and print the cost breakdown

Your structured logging system emits JSON entries for every agent action. The token tracker accumulates usage across the entire run. Let's examine both and confirm that token optimization is working.

  • Run the pipeline one more time with a fresh question:
python main.py
  • Enter a new business question:
Should a mid-size SaaS company invest in building internal AI agents?
  • When the human approval prompt appears, type yes and press Enter.
  • Scroll up through your terminal output and examine the structured JSON log lines. Each log entry includes timestamp, agent, action, tokens, and latency_ms fields.

How does token optimization work here?

The researcher agent collects raw search results from Tavily, then compresses them into a concise briefing before passing to the analyst. The raw_findings field stores up to 3,000 characters of raw data, while compressed_findings contains the summarized version.

This means the analyst and report writer only process the compressed context. The downstream agents never see the full raw search results. This architectural decision reduces token usage significantly across the pipeline.

  • Look at the --- Token Usage Summary --- at the bottom of the output. You should see a breakdown like:
--- Token Usage Summary ---
  planner: ~200 in / ~150 out ($0.0020)
  researcher: ~2000 in / ~400 out ($0.0090)
  analyst: ~800 in / ~300 out ($0.0050)
  report_writer: ~1200 in / ~600 out ($0.0090)
  TOTAL COST: $0.0250

Why is the researcher the most expensive agent?

The researcher processes large raw search results in its compression prompt. It sends the most input tokens because it includes up to 8,000 characters of raw Tavily output in the summarization call. This is exactly why token optimization matters. Without compression, the analyst and report writer would each re-process that full raw context.

  • Compare the researcher's token count against the analyst's. The analyst receives only the compressed findings, so its input token count should be significantly lower than the researcher's.
  • This difference proves the token optimization layer is working. Each agent passes only what the next agent needs.

Seeing 0 tokens for all agents?

If the token tracker shows zeros, the response.response_metadata might not include token_usage. Verify that your langchain-openai version is 1.3.3 by running pip show langchain-openai in your terminal. Older versions may use a different metadata path.

help me debug why token usage shows as 0 in my LangGraph pipeline using langchain-openai

Your multi-agent pipeline is fully validated. You have confirmed end-to-end execution with human-in-the-loop approval, verified observability through LangSmith traces, proven your guardrails catch invalid inputs, and measured token optimization with a real cost breakdown.

Secret mission

Add an Automated Evaluation Harness

Your pipeline generates intelligence reports, but how do you know if they're any good? In production, teams use LLM-as-judge evaluation to automatically score outputs on multiple dimensions. Build an evaluation harness that grades your reports on factual grounding, completeness, actionability, and coherence, then logs the results to LangSmith as structured feedback.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

This project runs entirely locally and charges per API call only when you execute the pipeline. There are no ongoing costs from leaving the files on your machine.

Resources you used:

  • business-intel-pipeline project folder on your Desktop (all source code, virtual environment, and .env file).
  • LangSmith business-intel-pipeline project traces (stored in your LangSmith account).
  • Tavily API key (free tier, stored in your Tavily Dashboard).
  • LangSmith account (free tier, holds your tracing data).

Keep everything running

No action needed. Choose this if you plan to keep experimenting with your multi-agent pipeline or want to try new business questions.

  • Your project folder stays on your Desktop, ready to run whenever you want.
  • Your Tavily account remains on the free tier (1,000 credits/month) with no charges.
  • Your LangSmith account remains free and stores your trace history.
  • OpenAI only charges when you actually run the pipeline. No idle costs.

Pause - I'll come back to this later

Deactivate your virtual environment to free up your terminal session. Your files and accounts stay intact for when you return.

  • Deactivate your virtual environment by running:
deactivate
  • No other action needed. Nothing is running in the background and no charges accrue while idle.
  • When you return, reactivate the virtual environment by navigating to your project folder and running source venv/bin/activate (macOS/Linux) or venv\Scripts\activate (Windows).

Delete - I don't want to use this again

Remove all project files and clean up cloud data. This is permanent.

  • Delete the project folder from your Desktop by running:

macOS

rm -rf ~/Desktop/business-intel-pipeline

Windows

rmdir /s /q %USERPROFILE%\Desktop\business-intel-pipeline
  • Delete your LangSmith project traces. Open your LangSmith projects list at smith.langchain.com, find the business-intel-pipeline project, click the overflow menu (three dots), and select Delete. This is permanent and cannot be undone.
  • Delete your Tavily API key. Open the Tavily Dashboard, navigate to the API Keys section, and click the trashcan icon next to your key.
  • Your LangSmith and Tavily accounts are both free tier with no charges. You can leave them active or close them via each service's account settings page.

Nice Work!

Nice Work!

You just built a production-grade multi-agent intelligence system from scratch. Your pipeline takes a natural language business question, orchestrates 4 specialized agents through a LangGraph StateGraph, and produces a structured intelligence brief with full observability.

You've learned how to:

  • Build a multi-agent pipeline with 4 specialized agents (Planner, Researcher, Analyst, Report Writer) orchestrated via a LangGraph StateGraph with conditional routing and loop control.
  • Implement human-in-the-loop approval using LangGraph's interrupt() and Command(resume=...) pattern with per-agent cost tracking and structured JSON logging.
  • Add production guardrails for input/output validation, a custom exception hierarchy, and full LangSmith tracing with visible tool calls and decision chains.
  • Secret Mission: Built an LLM-as-judge evaluation harness that scores report quality on 4 dimensions and logs metrics to LangSmith as feedback.

Ready to quiz yourself?