dailytutorfor.you
& AI Science Data

Building a Multi-Agent AI Workflow with LangGraph + FastAPI (Complete 2026 Guide)

A comprehensive Indonesian tutorial translated to English for building a production-minded multi-agent workflow using LangGraph, FastAPI, and Redis. Covers architecture, runnable implementation, best practices, common mistakes, and advanced tips.

9 min read

Building a Multi-Agent AI Workflow with LangGraph + FastAPI (Complete 2026 Guide)

1) Introduction — What & Why

If you actively follow 2026 developer trends, you’ve probably noticed one major pattern: agentic architecture is rising fast.

From quick research this morning:

  • GitHub Trending is full of repos about agent orchestration, multi-agent frameworks, and workflow automation (for example: the LangGraph ecosystem, Deep Agents discussions, and security tooling for AI pipelines).
  • On X Explore, AI topics are still dominant (OpenAI news, new tooling, and discussions about AI-generated code performance/quality).
  • On DEV Community, popular articles are usually about:
    • multi-agent deployment,
    • AI coding workflows,
    • agent security,
    • and “how to build production-ready AI systems”.

The problem is that many tutorials stop at a “hello world chatbot” demo. In reality, we need systems that:

  • can be split into multiple roles (planner, researcher, writer, reviewer),
  • are stateful (have process memory),
  • are observable,
  • and are safe to run in backend/API environments.

In this tutorial, we’ll build a truly runnable multi-agent workflow using:

  • LangGraph for graph orchestration,
  • FastAPI for exposing endpoints,
  • Redis for simple cache + memory,
  • plus practical error handling and a maintainable code structure.

Target: by the end, you’ll have an architecture foundation you can use for content generation, ticket triage, support automation, and internal copilots.

2) Prerequisites

Before starting, prepare:

Technical skills

  • Basic Python (functions, classes, typing, basic async)
  • Understanding of HTTP APIs (GET/POST)
  • Familiarity with the command line

Tools

  • Python 3.11+
  • pip / uv
  • Redis (local via Docker or service)
  • Editor (VS Code, Cursor, etc.)

Optional (for production)

  • Model provider API key (OpenAI/Anthropic/Gemini, etc.)
  • Logging/observability stack (e.g., LangSmith / OpenTelemetry)

3) Core Concepts (with analogy)

Imagine you have a small office team:

  • Planner = PM who breaks down the problem
  • Researcher = analyst who gathers facts
  • Writer = executor who produces output
  • Reviewer = QA who checks quality

Now, LangGraph is like your team’s SOP + flowchart.

Main concepts

  1. State

    • A “shared work board” where all agents read/write data.
    • Example: topic, outline, draft, warnings, final output.

  2. Node

    • One work unit (agent function).
    • For example planner_node, research_node, writer_node, reviewer_node.

  3. Edge

    • Transition path between nodes.
    • Can be linear, branching, or conditional.

  4. Durable mindset

    • Complex workflows often fail mid-run (timeout, API error, invalid data).
    • Design from the start with retry, fallback, and observability.

  5. Context control

    • One reason multi-agent systems are popular: preventing context bloat.
    • Subtasks are separated so all details don’t pile up into one huge prompt/state.

4) Architecture / Diagram

We’ll use a simple but realistic architecture:

Client (Web/App)
      |
      v
+-------------------+
| FastAPI Endpoint  |  POST /generate
+-------------------+
      |
      v
+-------------------+        +------------------+
| LangGraph Engine  | <----> | Redis Cache/State|
+-------------------+        +------------------+
      |
      v
[planner] -> [researcher] -> [writer] -> [reviewer]
      |                                          |
      +------------------(if fail)--------------+
                         retry/fix

Brief data flow

  1. Client sends a request (topic, audience, tone).
  2. FastAPI validates input with Pydantic.
  3. The workflow graph runs.
  4. Each node updates state.
  5. The final result is saved to Redis + returned to the client.

5) Step-by-Step Implementation (complete runnable code)

5.1 Project structure

multi-agent-workflow/
├── app/
│   ├── main.py
│   ├── graph_workflow.py
│   ├── schemas.py
│   └── storage.py
├── requirements.txt
└── README.md

5.2 Install dependencies

python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate
pip install -r requirements.txt

requirements.txt

fastapi==0.116.1
uvicorn==0.35.0
langgraph==0.3.5
pydantic==2.12.5
redis==6.2.0

5.3 Code: app/schemas.py

from pydantic import BaseModel, Field, field_validator


class GenerateRequest(BaseModel):
    topic: str = Field(..., min_length=5, max_length=200)
    audience: str = Field(default="beginner developer")
    tone: str = Field(default="practical")

    @field_validator("topic")
    @classmethod
    def topic_must_not_be_empty(cls, value: str) -> str:
        if not value.strip():
            raise ValueError("Topic must not be empty")
        return value.strip()


class GenerateResponse(BaseModel):
    request_id: str
    topic: str
    final_output: str
    quality_score: int
    warnings: list[str]

5.4 Code: app/storage.py

import json
from redis import Redis
from redis.exceptions import RedisError


class Storage:
    def __init__(self, url: str = "redis://localhost:6379/0"):
        self.client = Redis.from_url(url, decode_responses=True)

    def save_state(self, key: str, data: dict) -> None:
        """Save state to Redis with clear error handling."""
        try:
            self.client.setex(key, 3600, json.dumps(data, ensure_ascii=False))
        except RedisError as err:
            # Don’t crash the whole workflow just because caching fails
            print(f"[WARN] Failed to save to Redis: {err}")

    def load_state(self, key: str) -> dict | None:
        try:
            raw = self.client.get(key)
            return json.loads(raw) if raw else None
        except (RedisError, json.JSONDecodeError) as err:
            print(f"[WARN] Failed to read from Redis: {err}")
            return None

5.5 Code: app/graph_workflow.py

from __future__ import annotations

from typing import TypedDict
from langgraph.graph import StateGraph, START, END


class WorkflowState(TypedDict, total=False):
    request_id: str
    topic: str
    audience: str
    tone: str
    outline: str
    research_notes: str
    draft: str
    final_output: str
    quality_score: int
    warnings: list[str]


def planner_node(state: WorkflowState) -> WorkflowState:
    topic = state["topic"]
    outline = (
        f"Outline for topic '{topic}':
"
        "1) Definition
"
        "2) Architecture
"
        "3) Implementation
"
        "4) Best practices
"
        "5) Risks
"
    )
    return {"outline": outline}


def researcher_node(state: WorkflowState) -> WorkflowState:
    topic = state["topic"].lower()

    # Simulated research output (runnable without external APIs)
    notes = [
        "Use role separation across agents so context does not pile up.",
        "Add input-output validation between nodes.",
        "Provide retry + fallback when a node fails.",
        "Use cache for expensive subtask results.",
    ]

    if "security" in topic:
        notes.append("Sanitize prompt/tool input to prevent injection.")

    return {"research_notes": "
".join(f"- {n}" for n in notes)}


def writer_node(state: WorkflowState) -> WorkflowState:
    draft = (
        f"Topic: {state['topic']}

"
        f"Target audience: {state['audience']}
"
        f"Tone: {state['tone']}

"
        f"{state['outline']}
"
        "Research notes:
"
        f"{state['research_notes']}

"
        "Implementation starts from state design, then nodes, then observability."
    )
    return {"draft": draft}


def reviewer_node(state: WorkflowState) -> WorkflowState:
    draft = state["draft"]
    warnings: list[str] = []
    score = 90

    if len(draft) < 300:
        warnings.append("Draft too short")
        score -= 20

    if "error handling" not in draft.lower():
        warnings.append("Error handling is not emphasized yet")
        score -= 10

    final_output = draft + "

[Reviewed] Add logging, metrics, and tests before production."

    return {
        "final_output": final_output,
        "quality_score": max(score, 0),
        "warnings": warnings,
    }


def build_graph():
    graph = StateGraph(WorkflowState)

    graph.add_node("planner", planner_node)
    graph.add_node("researcher", researcher_node)
    graph.add_node("writer", writer_node)
    graph.add_node("reviewer", reviewer_node)

    graph.add_edge(START, "planner")
    graph.add_edge("planner", "researcher")
    graph.add_edge("researcher", "writer")
    graph.add_edge("writer", "reviewer")
    graph.add_edge("reviewer", END)

    return graph.compile()

5.6 Code: app/main.py

import uuid
from fastapi import FastAPI, HTTPException
from pydantic import ValidationError

from app.schemas import GenerateRequest, GenerateResponse
from app.graph_workflow import build_graph
from app.storage import Storage


app = FastAPI(title="Multi-Agent Workflow API", version="1.0.0")
workflow = build_graph()
storage = Storage()


@app.get("/health")
def health_check():
    return {"status": "ok"}


@app.post("/generate", response_model=GenerateResponse)
def generate_content(payload: GenerateRequest):
    request_id = str(uuid.uuid4())

    initial_state = {
        "request_id": request_id,
        "topic": payload.topic,
        "audience": payload.audience,
        "tone": payload.tone,
    }

    try:
        result = workflow.invoke(initial_state)

        # Save result for basic audit/debug
        storage.save_state(f"workflow:{request_id}", result)

        return GenerateResponse(
            request_id=request_id,
            topic=payload.topic,
            final_output=result.get("final_output", ""),
            quality_score=result.get("quality_score", 0),
            warnings=result.get("warnings", []),
        )

    except ValidationError as err:
        raise HTTPException(status_code=422, detail=str(err))
    except Exception as err:
        # Global error handling so API stays informative
        raise HTTPException(
            status_code=500,
            detail=f"Workflow processing failed: {err}",
        )

5.7 Run project

uvicorn app.main:app --reload --port 8000

Test endpoint:

curl -X POST http://127.0.0.1:8000/generate \
  -H 'Content-Type: application/json' \
  -d '{
    "topic": "Building a multi-agent AI for support tickets",
    "audience": "backend engineer",
    "tone": "practical"
  }'

If successful, you’ll get a JSON response containing final_output, quality_score, and warnings.

6) Best Practices (industry tips)

  1. Separate agent domains clearly

    • Don’t make every node “do everything”.
    • Task descriptions should be specific (planner plans, reviewer evaluates).

  2. Treat state as a contract

    • Treat state structure like an API contract.
    • Version your state as the workflow grows.

  3. Add guardrails in every node

    • Validate node input.
    • Limit output length.
    • Normalize format (JSON schema if needed).

  4. Observability is mandatory, not optional

    • Save traces of which node failed.
    • Record latency per node.
    • Log prompt/tool calls (without leaking secrets).

  5. Idempotency for production endpoints

    • If users retry requests, don’t always regenerate from scratch.
    • Use request fingerprints + cached results.

  6. Fail soft, not fail silent

    • When Redis is down, workflow should continue with in-memory fallback.
    • When one tool fails, try a fallback strategy.

7) Common Mistakes (and how to avoid them)

Mistake 1: “Multi-agent” but actually monolithic

Symptom: all logic sits in one large node.
Impact: hard to debug, context bloat, difficult scaling.
Solution: split nodes by responsibility.

Mistake 2: No schema validation

Symptom: inconsistent output between nodes.
Impact: random failures in the middle of workflow.
Solution: strict validation with Pydantic/TypedDict.

Mistake 3: Ignoring timeout/retry

Symptom: requests hang too long.
Impact: poor UX, higher API cost.
Solution: set explicit timeouts and limited retries (exponential backoff).

Mistake 4: Going to production without testing failure paths

Symptom: happy path works, edge cases collapse.
Impact: incidents become hard to diagnose.
Solution: test failed API scenarios, cache outages, invalid output.

Mistake 5: Ignoring prompt/tool injection

Symptom: agent follows dangerous instructions from input/tool results.
Impact: data leakage / unsafe actions.
Solution: sanitize inputs, enforce policy, allowlist tool actions.

8) Advanced Tips

If the foundation above is already stable, these upgrade paths are worth it:

  1. Conditional routing

    • Add branching edges: technical topics go to researcher A, business topics to researcher B.

  2. Parallel sub-agents

    • Research sources A/B/C in parallel, then aggregate.
    • This can reduce end-to-end latency.

  3. Human-in-the-loop approval

    • For sensitive steps (e.g., auto-publishing), insert an “approval required” node.

  4. Score-based quality gates

    • If quality_score < 80, automatically loop back to writer for revision.

  5. Hybrid memory strategy

    • Redis for short-term cache.
    • DB (Postgres) for audit trail and long-term memory.

  6. Model specialization

    • Planner uses a fast/cheap model.
    • Reviewer uses a stronger model for quality evaluation.

  7. Security hardening

    • Keep secrets in vault/env manager.
    • Rate-limit endpoints.
    • Redact sensitive data in input/output.

9) Summary & Next Steps

We just built a multi-agent system that:

  • has a clear architecture,
  • runs locally,
  • includes error handling,
  • and is ready to become a production foundation.

Key takeaways:

  • LangGraph is suitable for orchestrating complex state-based workflows.
  • FastAPI makes it easy to expose a fast, structured service.
  • Redis helps with cache/memory for performance and reliability.

Next steps I recommend

  1. Add automated tests (unit + integration).
  2. Add tracing/metrics for each node.
  3. Implement retry policy + circuit breaker.
  4. Connect to a real LLM provider (with model fallback).
  5. Add CI/CD + security scanning before production.

If you’re serious about building AI systems this year, learning orchestration + reliability + security will be far more valuable than just “prompting”.

10) References

Closing: The 2026 trend is clear: engineering teams are moving from single-agent demos to operable multi-agent systems. Hopefully this tutorial not only helps you “understand the concept,” but also gives you an implementation baseline you can evolve into production 🚀