FinGuard AI: Async Dispute Triage Agent

Build an async LangGraph agent backend that triages fintech disputes with LLM workflows.

Introduction

30 Second Summary

When you dispute a charge on your credit card, someone has to decide whether it's legitimate fraud or a misunderstanding. Banks process millions of these disputes daily, and the faster they triage them, the faster customers get resolution.

In this project, you will build FinGuard AI, an async backend that uses an LLM-powered agent workflow to automatically classify, investigate, and decide on transaction disputes.

What You'll Build

A FastAPI backend service that receives dispute payloads via a REST API, runs them through a 3-node LangGraph workflow powered by OpenAI's gpt-4o-mini, and returns triage decisions with full observability tracing.

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

  • A FastAPI async REST API with strict Pydantic request/response schemas for dispute intake.
  • A LangGraph StateGraph workflow that classifies, investigates, and decides on disputes using LiteLLM and OpenAI's gpt-4o-mini.
  • A fully containerized stack (Docker Compose with PostgreSQL and Redis) plus Langfuse tracing for end-to-end observability.
  • Secret Mission: Add a conditional edge that short-circuits high-value disputes directly to auto-escalation, bypassing the investigation node.

Are there any prerequisites?

This project assumes you have Docker and uv installed on your machine, along with Python 3.11 or higher. You'll also need an OpenAI API key and a free Langfuse Cloud account.

Familiarity with async Python and basic LangGraph concepts will help, but the project walks through each piece step by step.

Set Up the Project and Docker Compose

Before you can build the dispute triage agent, you need a solid project foundation. The FinGuard AI backend relies on multiple services working together: a FastAPI application server, a PostgreSQL database, and a Redis cache.

In this step, you will scaffold the project using uv (a fast Python package manager), define all dependencies, and write the Docker Compose configuration that orchestrates your multi-service stack with a single command.

In this step, get ready to:
  • Scaffold the project structure and configure all Python dependencies.
  • Write a Docker Compose file with three services (app, PostgreSQL, Redis).
  • Create a Dockerfile and verify the stack starts successfully.
Scaffold the project with uv

The uv tool handles Python dependency management and virtual environments. You will use it to create a minimal project skeleton, then fill in the full dependency list and tooling configuration.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir finguard-ai
cd finguard-ai
  • Initialize a bare uv project by running:
uv init --bare

Why --bare?

The --bare flag creates only a pyproject.toml without sample files, READMEs, or git initialization. Since you will define your own project structure, this gives you full control.

  • Create the src/ and tests/ directories with their Python package markers by running:
mkdir -p src tests
touch src/__init__.py tests/__init__.py
  • Confirm the directory structure by running:
ls

You should see pyproject.toml, src, and tests in the output.

Now you need to replace the generated pyproject.toml with your full dependency list and tool configuration. This file declares every package the project needs, plus Ruff linting rules and Pytest settings.

  • Open pyproject.toml in your text editor and replace its entire contents with:
[project]
name = "finguard-ai"
version = "0.1.0"
description = "Async fintech dispute triage agent backend"
requires-python = ">=3.11"
dependencies = [
    "fastapi[standard]==0.139.0",
    "langgraph==1.2.8",
    "litellm==1.91.1",
    "langfuse==2.59.7",
    "asyncpg==0.31.0",
    "redis>=5.0.0,<6.0.0",
    "pydantic>=2.10.0,<3.0.0",
    "python-dotenv>=1.0.0,<2.0.0",
]

[dependency-groups]
dev = [
    "pytest>=9.1.0,<10.0.0",
    "httpx>=0.28.0,<1.0.0",
    "ruff>=0.15.0",
]

[tool.ruff]
target-version = "py311"
line-length = 88

[tool.ruff.lint]
select = ["E", "F", "I", "N", "W"]

[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"

What does this configuration do?

The [project] section pins your core dependencies: FastAPI for the web framework, LangGraph for agent workflow orchestration, LiteLLM for provider-agnostic LLM calls, and Langfuse for observability tracing.

The [dependency-groups] section separates dev tools (pytest, httpx for test HTTP requests, and Ruff for linting) from production dependencies. The [tool.ruff] section configures code quality rules targeting Python 3.11.

  • Save pyproject.toml.

Seeing a TOML syntax error?

Make sure all quotes are straight double quotes " and not curly/smart quotes. Also check that every opening bracket [ has a matching close ].

Help me debug my pyproject.toml

Next, create a .env.example file inside src/ so team members know which environment variables are required.

  • Create src/.env.example with the following content:
OPENAI_API_KEY=your-openai-api-key-here
LANGFUSE_SECRET_KEY=your-langfuse-secret-key-here
LANGFUSE_PUBLIC_KEY=your-langfuse-public-key-here
LANGFUSE_BASE_URL=https://cloud.langfuse.com

What is .env.example for?

This file documents the environment variables the app needs without storing real secrets. Each team member copies it to their own .env and fills in real values.

  • Save src/.env.example.
  • Generate the lock file from your dependency list by running:
uv lock

You should see uv resolve all packages and create a uv.lock file in your project root.

Lock resolution failed?

Check that your requires-python is set to ">=3.11" and that your system Python meets this requirement. Also verify you have internet access for downloading package metadata.

Help me fix my uv lock error

✔️ Awesome, I've got everything!

Great. Double check that you have saved pyproject.toml and src/.env.example.

ⓧ I'd like to double check the full code

Here is the full pyproject.toml:

[project]
name = "finguard-ai"
version = "0.1.0"
description = "Async fintech dispute triage agent backend"
requires-python = ">=3.11"
dependencies = [
    "fastapi[standard]==0.139.0",
    "langgraph==1.2.8",
    "litellm==1.91.1",
    "langfuse==2.59.7",
    "asyncpg==0.31.0",
    "redis>=5.0.0,<6.0.0",
    "pydantic>=2.10.0,<3.0.0",
    "python-dotenv>=1.0.0,<2.0.0",
]

[dependency-groups]
dev = [
    "pytest>=9.1.0,<10.0.0",
    "httpx>=0.28.0,<1.0.0",
    "ruff>=0.15.0",
]

[tool.ruff]
target-version = "py311"
line-length = 88

[tool.ruff.lint]
select = ["E", "F", "I", "N", "W"]

[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"

Here is src/.env.example:

OPENAI_API_KEY=your-openai-api-key-here
LANGFUSE_SECRET_KEY=your-langfuse-secret-key-here
LANGFUSE_PUBLIC_KEY=your-langfuse-public-key-here
LANGFUSE_BASE_URL=https://cloud.langfuse.com
Write the Docker Compose file

Docker Compose lets you define and run multiple containers as a single stack. Your docker-compose.yml will declare three services: the FastAPI app, PostgreSQL for persistence, and Redis for caching.

  • Create a new file called docker-compose.yml in the finguard-ai project root with the following content:
services:
  app:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql+asyncpg://finguard:finguard@postgres:5432/finguard
      - REDIS_URL=redis://redis:6379/0
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - LANGFUSE_SECRET_KEY=${LANGFUSE_SECRET_KEY}
      - LANGFUSE_PUBLIC_KEY=${LANGFUSE_PUBLIC_KEY}
      - LANGFUSE_BASE_URL=https://cloud.langfuse.com
    depends_on:
      postgres:
        condition: service_healthy
      redis:
        condition: service_healthy

  postgres:
    image: postgres:16
    environment:
      - POSTGRES_USER=finguard
      - POSTGRES_PASSWORD=finguard
      - POSTGRES_DB=finguard
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U finguard"]
      interval: 5s
      timeout: 3s
      retries: 5

  redis:
    image: redis:7
    ports:
      - "6379:6379"
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 3s
      retries: 5

volumes:
  pgdata:

What does this Docker Compose file do?

The app service builds your FastAPI container from the local Dockerfile. It passes credentials through environment variables, using ${OPENAI_API_KEY} syntax to read from your shell or a .env file at the project root.

The postgres and redis services use official images and define health checks so the app only starts after both databases are ready. The pgdata volume persists database data between container restarts.

  • Save docker-compose.yml.

YAML indentation errors?

YAML is whitespace-sensitive. Make sure you use spaces (not tabs) and that nested keys are indented by exactly 2 spaces relative to their parent.

Help me fix my docker-compose.yml

✔️ Awesome, I've got everything!

Great. Make sure you have saved docker-compose.yml in the project root.

ⓧ I'd like to double check the full code

services:
  app:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql+asyncpg://finguard:finguard@postgres:5432/finguard
      - REDIS_URL=redis://redis:6379/0
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - LANGFUSE_SECRET_KEY=${LANGFUSE_SECRET_KEY}
      - LANGFUSE_PUBLIC_KEY=${LANGFUSE_PUBLIC_KEY}
      - LANGFUSE_BASE_URL=https://cloud.langfuse.com
    depends_on:
      postgres:
        condition: service_healthy
      redis:
        condition: service_healthy

  postgres:
    image: postgres:16
    environment:
      - POSTGRES_USER=finguard
      - POSTGRES_PASSWORD=finguard
      - POSTGRES_DB=finguard
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U finguard"]
      interval: 5s
      timeout: 3s
      retries: 5

  redis:
    image: redis:7
    ports:
      - "6379:6379"
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 3s
      retries: 5

volumes:
  pgdata:
Create the Dockerfile and verify the stack

The Dockerfile defines how your FastAPI app is packaged into a container image. It uses a slim Python base image and copies uv directly from its official container image for fast dependency installation.

  • Create a file called Dockerfile (no extension) in the finguard-ai project root with the following content:
FROM python:3.11-slim

# Copy uv binary from its official image
COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/

WORKDIR /app

# Install dependencies from the lock file
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev

# Copy application source code
COPY src/ ./src/

EXPOSE 8000
CMD ["uv", "run", "uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8000"]

What does this Dockerfile do?

The COPY --from=ghcr.io/astral-sh/uv:latest line pulls the uv binary from its official container image in a multi-stage copy. This avoids installing uv via curl inside the container.

The --frozen flag on uv sync ensures the exact versions from uv.lock are installed without re-resolving. The --no-dev flag skips development dependencies (pytest, ruff) to keep the production image lean.

  • Save Dockerfile.

✔️ Awesome, I've got everything!

Great. Make sure Dockerfile is saved in the project root alongside docker-compose.yml.

ⓧ I'd like to double check the full code

FROM python:3.11-slim

COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/

WORKDIR /app

COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev

COPY src/ ./src/

EXPOSE 8000
CMD ["uv", "run", "uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8000"]

Before you can start the stack, Docker Compose needs your API keys. The ${OPENAI_API_KEY} references in docker-compose.yml read from a .env file at the project root.

  • Create a .env file in the finguard-ai root (same directory as docker-compose.yml) with your real credentials:
OPENAI_API_KEY=your-openai-api-key-here
LANGFUSE_SECRET_KEY=your-langfuse-secret-key-here
LANGFUSE_PUBLIC_KEY=your-langfuse-public-key-here
LANGFUSE_BASE_URL=https://cloud.langfuse.com
  • Replace each your-...-here placeholder with your actual key values from your OpenAI and Langfuse dashboards.
  • Save .env.

Never commit .env to git

Your .env file contains real secrets. If you initialize a git repo later, add .env to your .gitignore immediately.

Now verify that the infrastructure services (PostgreSQL and Redis) start correctly. The app service will fail at this point because src/main.py does not exist yet.

  • Start the stack by running:
docker compose up --build

You should see Docker pull the postgres:16 and redis:7 images, then show health check logs confirming both services are ready. The app container will exit with an error because src/main.py is missing. That is expected.

What should I see?

Look for lines containing postgres reporting "database system is ready to accept connections" and redis reporting "Ready to accept connections". These confirm your infrastructure is healthy.

  • Stop the stack by pressing Ctrl+C in the terminal, then run:
docker compose down

Docker daemon not running?

If you see an error about connecting to the Docker daemon, make sure Docker Desktop is running. Open Docker Desktop from your applications and wait for the engine status to show "Running".

Help me troubleshoot Docker Compose

Your project foundation is solid. You have a fully configured dependency manifest, a containerized infrastructure stack, and a reproducible build. Next up, you will build the Pydantic schemas and FastAPI endpoints that accept dispute payloads.

Define Schemas and FastAPI Endpoints

Your Docker stack is running, but it has no application logic yet. Before building the AI workflow, you need a clear contract that defines what data goes into the system and what comes out.

In this step, you will define Pydantic schemas for the dispute request and response, create a FastAPI application with a lifespan handler that connects to PostgreSQL and Redis, and wire up the API endpoints.

In this step, get ready to:
  • Define Pydantic request and response models with strict validation.
  • Create the FastAPI app with an async lifespan handler for database and cache connections.
  • Add a health check endpoint that verifies infrastructure connectivity.
Define the Pydantic request and response models

Every dispute that enters the system needs a consistent shape. Pydantic models enforce that shape at the API boundary, rejecting malformed payloads before they reach your workflow.

  • Create a new file called src/schemas.py in your editor.
  • Add the following code to define both models:
from pydantic import BaseModel, Field


# Defines the shape of incoming dispute data from the client
class DisputeRequest(BaseModel):
    transaction_id: str = Field(..., description="Unique transaction identifier")
    amount: float = Field(..., gt=0, description="Disputed amount in USD")
    merchant: str = Field(..., description="Merchant name")
    description: str = Field(..., description="Customer's dispute description")
    customer_id: str = Field(..., description="Customer identifier")


# Defines the shape of the triage decision returned to the client
class DisputeResponse(BaseModel):
    dispute_id: str = Field(..., description="Generated dispute case ID")
    classification: str = Field(..., description="Fraud type classification")
    decision: str = Field(..., description="Triage decision: approve, escalate, deny")
    reasoning: str = Field(..., description="Explanation of the decision")
    confidence_score: float = Field(
        ..., ge=0.0, le=1.0, description="Confidence in the decision"
    )

What does this code do?

DisputeRequest defines the five fields a client must send when submitting a dispute. The gt=0 constraint on amount ensures negative or zero values are rejected automatically.

DisputeResponse defines what the API sends back after triage. The ge=0.0, le=1.0 constraint on confidence_score guarantees the value stays between 0 and 1.

  • Save src/schemas.py.

You won't see a visible change yet. The next chunk uses these models to build the endpoint.

Create the FastAPI app with the lifespan handler

FastAPI's lifespan context manager lets you run setup code (like connecting to databases) when the app starts and teardown code when it shuts down. This keeps your connection pools managed in one place.

  • Create a new file called src/main.py and add the following code:
from contextlib import asynccontextmanager

import asyncpg
import redis.asyncio as aioredis
from fastapi import FastAPI

from src.routes import router
from src.tracing import get_langfuse_handler


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Create a PostgreSQL connection pool on startup
    app.state.pg_pool = await asyncpg.create_pool(
        dsn="postgresql://finguard:finguard@postgres:5432/finguard"
    )
    # Create a Redis client on startup
    app.state.redis = aioredis.from_url("redis://redis:6379/0")
    # Initialize the Langfuse tracing handler
    app.state.langfuse_handler = get_langfuse_handler()
    yield
    # Close connections on shutdown
    await app.state.pg_pool.close()
    await app.state.redis.close()


app = FastAPI(title="FinGuard AI", version="0.1.0", lifespan=lifespan)
app.include_router(router)

What does this code do?

The lifespan function runs once when the app boots. It creates an asyncpg connection pool for PostgreSQL and a redis.asyncio client, storing both on app.state so any endpoint can access them.

After yield, the teardown block runs when the server shuts down, cleanly closing both connections. This pattern prevents resource leaks.

  • Save src/main.py.

You won't see a change yet. The src/routes.py and src/tracing.py imports are created next.

Why store connections on app.state?

Storing the pool on app.state means every request handler gets access through request.app.state. This avoids creating a new database connection on every request, which would be slow and wasteful.

Add the triage and health check endpoints

The router defines two endpoints. The triage endpoint accepts dispute payloads and invokes the LangGraph workflow. The health check verifies that PostgreSQL and Redis are reachable.

  • Create a new file called src/routes.py and add the triage endpoint:
import uuid

from fastapi import APIRouter, Request

from src.schemas import DisputeRequest, DisputeResponse
from src.workflow import build_dispute_graph

router = APIRouter()


@router.post("/disputes/triage", response_model=DisputeResponse)
async def triage_dispute(request: Request, dispute: DisputeRequest):
    # Build the LangGraph workflow
    graph = build_dispute_graph()
    langfuse_handler = request.app.state.langfuse_handler

    # Prepare the initial state from the request payload
    initial_state = {
        "transaction_id": dispute.transaction_id,
        "amount": dispute.amount,
        "merchant": dispute.merchant,
        "description": dispute.description,
        "customer_id": dispute.customer_id,
        "classification": "",
        "decision": "",
        "reasoning": "",
        "confidence_score": 0.0,
    }

    # Invoke the graph asynchronously with Langfuse tracing
    result = await graph.ainvoke(
        initial_state, config={"callbacks": [langfuse_handler]}
    )

    # Return the final triage decision
    return DisputeResponse(
        dispute_id=str(uuid.uuid4()),
        classification=result["classification"],
        decision=result["decision"],
        reasoning=result["reasoning"],
        confidence_score=result["confidence_score"],
    )

What does this code do?

The triage_dispute function receives a DisputeRequest payload, builds the initial workflow state from it, and invokes the LangGraph graph asynchronously. The Langfuse callback handler is attached so every run is traced.

The build_dispute_graph() import references a module you will create in the next step. The endpoint is fully wired and ready to use once that module exists.

  • Now add the health check endpoint below the triage function in the same file:
@router.get("/health")
async def health_check(request: Request):
    # Retrieve connection references from app state
    pg_pool = request.app.state.pg_pool
    redis_client = request.app.state.redis

    # Test PostgreSQL connectivity
    pg_ok = False
    try:
        async with pg_pool.acquire() as conn:
            await conn.fetchval("SELECT 1")
        pg_ok = True
    except Exception:
        pass

    # Test Redis connectivity
    redis_ok = False
    try:
        await redis_client.ping()
        redis_ok = True
    except Exception:
        pass

    # Return overall status
    status = "healthy" if (pg_ok and redis_ok) else "degraded"
    return {"status": status, "postgres": pg_ok, "redis": redis_ok}

What does this code do?

The health check pulls the PostgreSQL pool and Redis client from app.state and attempts a lightweight query on each. If both succeed, the status is healthy. If either fails, it returns degraded instead of crashing.

This pattern lets monitoring tools (or you during development) confirm the infrastructure is reachable without invoking the full AI workflow.

  • Save src/routes.py.
  • Verify your code is lint-clean by running:
uv run ruff check src/schemas.py src/main.py src/routes.py

You should see no output if everything passes. Any output means Ruff found a style or syntax issue to fix.

Seeing lint errors?

If Ruff reports an import sorting issue, run uv run ruff check --fix src/ to auto-fix it. If you see an undefined name error, double-check that your import statements at the top of each file match the code above exactly.

help me fix the Ruff lint errors in my FinGuard AI project

Why can't I run the server yet?

The src/main.py file imports src.tracing and src/routes.py imports src.workflow. Those modules do not exist yet. You will create them in the next two steps, after which docker compose up --build will start the full app and both endpoints will respond.

✔️ Awesome, I've got everything!

Great. Double check you have saved all three files before moving on.

ⓧ I'd like to double check the full code

from pydantic import BaseModel, Field


class DisputeRequest(BaseModel):
    transaction_id: str = Field(..., description="Unique transaction identifier")
    amount: float = Field(..., gt=0, description="Disputed amount in USD")
    merchant: str = Field(..., description="Merchant name")
    description: str = Field(..., description="Customer's dispute description")
    customer_id: str = Field(..., description="Customer identifier")


class DisputeResponse(BaseModel):
    dispute_id: str = Field(..., description="Generated dispute case ID")
    classification: str = Field(..., description="Fraud type classification")
    decision: str = Field(..., description="Triage decision: approve, escalate, deny")
    reasoning: str = Field(..., description="Explanation of the decision")
    confidence_score: float = Field(
        ..., ge=0.0, le=1.0, description="Confidence in the decision"
    )
from contextlib import asynccontextmanager

import asyncpg
import redis.asyncio as aioredis
from fastapi import FastAPI

from src.routes import router
from src.tracing import get_langfuse_handler


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.pg_pool = await asyncpg.create_pool(
        dsn="postgresql://finguard:finguard@postgres:5432/finguard"
    )
    app.state.redis = aioredis.from_url("redis://redis:6379/0")
    app.state.langfuse_handler = get_langfuse_handler()
    yield
    await app.state.pg_pool.close()
    await app.state.redis.close()


app = FastAPI(title="FinGuard AI", version="0.1.0", lifespan=lifespan)
app.include_router(router)
import uuid

from fastapi import APIRouter, Request

from src.schemas import DisputeRequest, DisputeResponse
from src.workflow import build_dispute_graph

router = APIRouter()


@router.post("/disputes/triage", response_model=DisputeResponse)
async def triage_dispute(request: Request, dispute: DisputeRequest):
    graph = build_dispute_graph()
    langfuse_handler = request.app.state.langfuse_handler

    initial_state = {
        "transaction_id": dispute.transaction_id,
        "amount": dispute.amount,
        "merchant": dispute.merchant,
        "description": dispute.description,
        "customer_id": dispute.customer_id,
        "classification": "",
        "decision": "",
        "reasoning": "",
        "confidence_score": 0.0,
    }

    result = await graph.ainvoke(
        initial_state, config={"callbacks": [langfuse_handler]}
    )

    return DisputeResponse(
        dispute_id=str(uuid.uuid4()),
        classification=result["classification"],
        decision=result["decision"],
        reasoning=result["reasoning"],
        confidence_score=result["confidence_score"],
    )


@router.get("/health")
async def health_check(request: Request):
    pg_pool = request.app.state.pg_pool
    redis_client = request.app.state.redis

    pg_ok = False
    try:
        async with pg_pool.acquire() as conn:
            await conn.fetchval("SELECT 1")
        pg_ok = True
    except Exception:
        pass

    redis_ok = False
    try:
        await redis_client.ping()
        redis_ok = True
    except Exception:
        pass

    status = "healthy" if (pg_ok and redis_ok) else "degraded"
    return {"status": status, "postgres": pg_ok, "redis": redis_ok}

Your API contract is locked in. Next up, you will build the LangGraph workflow that powers the triage logic behind the /disputes/triage endpoint.

Build the LangGraph Dispute Workflow

Your FastAPI endpoints are wired up and your Docker stack is running. The /disputes/triage endpoint exists, but it references a workflow module that does not exist yet.

In this step, you will build the brain of FinGuard AI. A LangGraph StateGraph orchestrates three async nodes. Each node calls an LLM to classify the dispute, investigate risk factors, and render a triage decision.

In this step, get ready to:
  • Define the workflow state schema and implement the LLM-powered node functions.
  • Wire the nodes into a compiled StateGraph with linear edges.
  • Create the Langfuse tracing module and verify the full pipeline end-to-end.
Define the workflow state and node functions

The workflow state is a Python TypedDict that flows through every node in the graph. Each node reads from the state, calls the LLM, and returns a partial update.

  • Create a new file src/workflow.py inside your finguard-ai project folder.
  • Add the imports and the DisputeState TypedDict by pasting the following code:
from typing import TypedDict

from langgraph.graph import END, START, StateGraph
from litellm import acompletion


class DisputeState(TypedDict):
    transaction_id: str
    amount: float
    merchant: str
    description: str
    customer_id: str
    classification: str
    decision: str
    reasoning: str
    confidence_score: float

What does this code do?

The imports pull in StateGraph, START, and END from LangGraph for graph construction. acompletion is LiteLLM's async completion function that talks to OpenAI.

DisputeState defines every field that passes between nodes. Nodes read from this state and return partial dictionaries to update specific keys.

  • Save src/workflow.py.
  • Confirm the file exists by running:
ls src/workflow.py

You should see src/workflow.py in the output.

File not found?

Make sure you created the file inside the src/ directory (not the project root). Your terminal should be inside the finguard-ai folder.

help me find where to create src/workflow.py in my project

  • Add the first node function, classify_dispute, below the DisputeState class:
async def classify_dispute(state: DisputeState) -> dict:
    prompt = (
        f"Classify this transaction dispute into one of these categories: "
        f"unauthorized_charge, duplicate_charge, product_not_received, "
        f"product_not_as_described, billing_error.\n\n"
        f"Transaction: ${state['amount']} at {state['merchant']}\n"
        f"Customer says: {state['description']}\n\n"
        f"Respond with ONLY the category name."
    )
    response = await acompletion(
        model="openai/gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
    )
    classification = response.choices[0].message.content.strip()
    return {"classification": classification}

What does this node do?

classify_dispute builds a structured prompt with the dispute details and asks the LLM to return one of five fraud categories. The openai/gpt-4o-mini model string uses LiteLLM's provider-prefix format so LiteLLM routes the call to the correct API.

The function returns a dictionary with only the classification key. LangGraph merges this partial update into the shared state automatically.

  • Save src/workflow.py. You will not see a visible change yet. The next two functions build on the classification this node produces.
  • Add the investigate_dispute function below classify_dispute:
async def investigate_dispute(state: DisputeState) -> dict:
    prompt = (
        f"You are a fraud analyst. Investigate this dispute:\n"
        f"Type: {state['classification']}\n"
        f"Amount: ${state['amount']} at {state['merchant']}\n"
        f"Customer says: {state['description']}\n\n"
        f"Provide a brief 2-sentence analysis of risk factors."
    )
    response = await acompletion(
        model="openai/gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
    )
    reasoning = response.choices[0].message.content.strip()
    return {"reasoning": reasoning}

What does this node do?

investigate_dispute reads the classification that the previous node added to the state. It asks the LLM to act as a fraud analyst and return a concise risk analysis.

The returned reasoning string feeds into the final decision node.

  • Save src/workflow.py. No visible change yet. The next function uses both classification and reasoning to render the final decision.
  • Add the decide_dispute function below investigate_dispute:
async def decide_dispute(state: DisputeState) -> dict:
    prompt = (
        f"Based on this investigation, render a triage decision.\n"
        f"Classification: {state['classification']}\n"
        f"Analysis: {state['reasoning']}\n"
        f"Amount: ${state['amount']}\n\n"
        f"Respond in this exact format:\n"
        f"DECISION: approve|escalate|deny\n"
        f"CONFIDENCE: 0.0-1.0"
    )
    response = await acompletion(
        model="openai/gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
    )
    content = response.choices[0].message.content.strip()

    decision = "escalate"
    confidence = 0.5
    for line in content.split("\n"):
        if line.startswith("DECISION:"):
            decision = line.split(":")[1].strip().lower()
        elif line.startswith("CONFIDENCE:"):
            try:
                confidence = float(line.split(":")[1].strip())
            except ValueError:
                confidence = 0.5

    return {"decision": decision, "confidence_score": confidence}

What does this node do?

decide_dispute asks the LLM to choose approve, escalate, or deny and return a confidence score. The parsing logic extracts both values from the structured response.

Fallback defaults ("escalate" and 0.5) protect against malformed LLM output. In production, you always want safe defaults when parsing free-text responses.

  • Save src/workflow.py. The three node functions are complete. Next you will wire them into a graph.
Build the StateGraph with edges

A StateGraph is a builder object. You register nodes, connect them with edges, and then compile the builder into an executable graph.

  • Add the build_dispute_graph function at the bottom of src/workflow.py:
def build_dispute_graph() -> StateGraph:
    graph = StateGraph(DisputeState)

    graph.add_node("classify", classify_dispute)
    graph.add_node("investigate", investigate_dispute)
    graph.add_node("decide", decide_dispute)

    graph.add_edge(START, "classify")
    graph.add_edge("classify", "investigate")
    graph.add_edge("investigate", "decide")
    graph.add_edge("decide", END)

    return graph.compile()

What does this function do?

build_dispute_graph creates a linear pipeline: START goes to classify, classify goes to investigate, investigate goes to decide, and decide goes to END. The compile() call transforms the builder into an executable object that supports ainvoke() for async execution.

Your routes.py already calls build_dispute_graph() and invokes the compiled graph with graph.ainvoke(). Once this file exists, the import resolves.

  • Save src/workflow.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved src/workflow.py with all four functions (classify_dispute, investigate_dispute, decide_dispute, build_dispute_graph).

ⓧ I'd like to double check the full code

Here is the complete src/workflow.py file:

from typing import TypedDict

from langgraph.graph import END, START, StateGraph
from litellm import acompletion


class DisputeState(TypedDict):
    transaction_id: str
    amount: float
    merchant: str
    description: str
    customer_id: str
    classification: str
    decision: str
    reasoning: str
    confidence_score: float


async def classify_dispute(state: DisputeState) -> dict:
    prompt = (
        f"Classify this transaction dispute into one of these categories: "
        f"unauthorized_charge, duplicate_charge, product_not_received, "
        f"product_not_as_described, billing_error.\n\n"
        f"Transaction: ${state['amount']} at {state['merchant']}\n"
        f"Customer says: {state['description']}\n\n"
        f"Respond with ONLY the category name."
    )
    response = await acompletion(
        model="openai/gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
    )
    classification = response.choices[0].message.content.strip()
    return {"classification": classification}


async def investigate_dispute(state: DisputeState) -> dict:
    prompt = (
        f"You are a fraud analyst. Investigate this dispute:\n"
        f"Type: {state['classification']}\n"
        f"Amount: ${state['amount']} at {state['merchant']}\n"
        f"Customer says: {state['description']}\n\n"
        f"Provide a brief 2-sentence analysis of risk factors."
    )
    response = await acompletion(
        model="openai/gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
    )
    reasoning = response.choices[0].message.content.strip()
    return {"reasoning": reasoning}


async def decide_dispute(state: DisputeState) -> dict:
    prompt = (
        f"Based on this investigation, render a triage decision.\n"
        f"Classification: {state['classification']}\n"
        f"Analysis: {state['reasoning']}\n"
        f"Amount: ${state['amount']}\n\n"
        f"Respond in this exact format:\n"
        f"DECISION: approve|escalate|deny\n"
        f"CONFIDENCE: 0.0-1.0"
    )
    response = await acompletion(
        model="openai/gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
    )
    content = response.choices[0].message.content.strip()

    decision = "escalate"
    confidence = 0.5
    for line in content.split("\n"):
        if line.startswith("DECISION:"):
            decision = line.split(":")[1].strip().lower()
        elif line.startswith("CONFIDENCE:"):
            try:
                confidence = float(line.split(":")[1].strip())
            except ValueError:
                confidence = 0.5

    return {"decision": decision, "confidence_score": confidence}


def build_dispute_graph() -> StateGraph:
    graph = StateGraph(DisputeState)

    graph.add_node("classify", classify_dispute)
    graph.add_node("investigate", investigate_dispute)
    graph.add_node("decide", decide_dispute)

    graph.add_edge(START, "classify")
    graph.add_edge("classify", "investigate")
    graph.add_edge("investigate", "decide")
    graph.add_edge("decide", END)

    return graph.compile()
Create the tracing module and verify the integration

Your src/main.py already imports get_langfuse_handler from src.tracing. You need to create that module so the app starts without an import error.

  • Create a new file src/tracing.py and paste the following code:
import os

from langfuse.langchain import CallbackHandler


def get_langfuse_handler() -> CallbackHandler:
    os.environ.setdefault("LANGFUSE_SECRET_KEY", "")
    os.environ.setdefault("LANGFUSE_PUBLIC_KEY", "")
    os.environ.setdefault("LANGFUSE_BASE_URL", "https://cloud.langfuse.com")
    return CallbackHandler()

What does this module do?

get_langfuse_handler initializes a Langfuse CallbackHandler that LangGraph uses to emit traces. The setdefault calls ensure the app does not crash if the environment variables are missing. It just runs without tracing in that case.

When you pass this handler to graph.ainvoke() via the config={"callbacks": [langfuse_handler]} pattern, every node execution appears as a span in the Langfuse dashboard.

  • Save src/tracing.py.
  • Rebuild and start the Docker stack by running:
docker compose up --build
  • Wait for the app container to log that it is listening on port 8000.
  • Open a second terminal and send a test dispute by running:
curl -X POST http://localhost:8000/disputes/triage \
  -H "Content-Type: application/json" \
  -d '{"transaction_id":"txn_001","amount":499.99,"merchant":"ElectroMart","description":"I did not authorize this purchase","customer_id":"cust_123"}'

You should see a JSON response containing dispute_id, classification, decision, reasoning, and confidence_score. The decision will be one of approve, escalate, or deny.

Seeing a connection error or import failure?

  • If you see an ImportError mentioning src.workflow or src.tracing, double check that both files are saved inside the src/ directory and that src/__init__.py exists (it can be empty).
  • If the response hangs or times out, verify your OPENAI_API_KEY is set in your .env file. The LiteLLM call needs a valid key to reach the OpenAI API.
  • If the health check at /health returns degraded, wait a few seconds for PostgreSQL and Redis to finish their health checks before retrying.

help me debug why my FinGuard AI dispute triage endpoint is failing

Your dispute triage pipeline is live. In the next step, you will configure Langfuse tracing so every dispute creates a visible trace with token usage and latency per node.

Add Langfuse Tracing and Observability

In the previous step, you built a 3-node LangGraph workflow that classifies, investigates, and decides on disputes. The workflow runs, but you have no visibility into what the LLM actually did at each step.

Observability solves this. In this step, you will configure Langfuse to capture a full trace of every dispute that flows through the pipeline, showing you token usage, latency, and the exact prompts at each node.

In this step, get ready to:
  • Configure Langfuse credentials for the Docker stack.
  • Understand how the CallbackHandler attaches to graph invocations.
  • Send a test dispute and verify the trace appears in the Langfuse dashboard.
Configure Langfuse credentials via environment variables

Langfuse Cloud is a hosted observability platform that stores traces from your LLM applications. Your docker-compose.yml already references three Langfuse environment variables, but they are not populated yet.

  • Open Langfuse Cloud at cloud.langfuse.com and sign in to your account.
  • Create a new project (name it anything you like, such as finguard-ai).
  • Navigate to your project's Settings, then select API Keys.
  • Copy the Secret Key (prefixed sk-lf-) and the Public Key (prefixed pk-lf-).

Why two keys?

The public key identifies your project. The secret key authenticates write access. Together they let the Langfuse SDK send trace data to the correct project without exposing full account credentials.

  • In your code editor, create a new file called .env in the project root (the same directory as docker-compose.yml) with these values:
OPENAI_API_KEY=your-openai-api-key-here
LANGFUSE_SECRET_KEY=your-langfuse-secret-key-here
LANGFUSE_PUBLIC_KEY=your-langfuse-public-key-here
LANGFUSE_BASE_URL=https://cloud.langfuse.com
  • Replace your-openai-api-key-here with your actual OpenAI API key.
  • Replace your-langfuse-secret-key-here with the secret key you copied.
  • Replace your-langfuse-public-key-here with the public key you copied.
  • Save the .env file.

How does Docker Compose pick these up?

Docker Compose automatically reads a .env file in the same directory as docker-compose.yml. The ${LANGFUSE_SECRET_KEY} syntax in your compose file substitutes the value from this file into the container's environment at startup.

Keys not working later?

Make sure there are no spaces around the = signs in your .env file. Also confirm you copied the full key value without trailing whitespace.

Help me debug my Langfuse environment variable configuration.

Review how the CallbackHandler attaches to graph invocations

The tracing code is already wired into your application from the previous steps. Let's walk through how the pieces connect so you understand the full flow.

Open src/tracing.py in your editor. This module initializes the Langfuse handler:

import os

from langfuse.langchain import CallbackHandler


def get_langfuse_handler() -> CallbackHandler:
    os.environ.setdefault("LANGFUSE_SECRET_KEY", "")
    os.environ.setdefault("LANGFUSE_PUBLIC_KEY", "")
    os.environ.setdefault("LANGFUSE_BASE_URL", "https://cloud.langfuse.com")
    return CallbackHandler()

What does this code do?

The CallbackHandler from langfuse.langchain is a LangChain-compatible callback that intercepts every node invocation in a LangGraph workflow. It records spans (one per node), token counts, latency, and prompt/response pairs.

The os.environ.setdefault calls ensure the environment variables exist (even if empty) before the handler initializes. At runtime inside Docker, the real values from your .env file are injected by Compose.

Now look at how the handler is used. In src/routes.py, find the triage_dispute function. The critical line is:

    result = await graph.ainvoke(
        initial_state, config={"callbacks": [langfuse_handler]}
    )

Why pass callbacks in config?

LangGraph's ainvoke() accepts a config dictionary that propagates to every node in the graph. By passing the Langfuse handler here, each node (classify, investigate, decide) automatically reports its execution as a span within a single trace.

This means you get a full trace per API request with zero changes to your node functions.

Run a test dispute and verify the trace in Langfuse

With credentials configured and the callback handler wired, you can now send a live dispute through the system and watch the trace appear in your Langfuse dashboard.

  • Rebuild and restart your Docker stack by running this command:
docker compose up --build -d

What does this command do?

The --build flag rebuilds the app image with any file changes. The -d flag runs the containers in detached mode so you get your terminal back.

  • Wait about 10 seconds for the health checks to pass, then send a test dispute by running this command:
curl -X POST http://localhost:8000/disputes/triage \
  -H "Content-Type: application/json" \
  -d '{"transaction_id": "txn_test_001", "amount": 499.99, "merchant": "ElectroMart", "description": "I did not authorize this purchase", "customer_id": "cust_123"}'

What should I see?

You should receive a JSON response containing a dispute_id, a classification (such as unauthorized_charge), a decision (approve, escalate, or deny), reasoning explaining the decision, and a confidence_score between 0 and 1.

Seeing a connection error or 500 response?

Check that your containers are running with docker compose ps. All three services (app, postgres, redis) should show a healthy status.

If the app container exited, check its logs with docker compose logs app. A missing or invalid OPENAI_API_KEY is the most common cause of 500 errors.

Help me debug my FinGuard AI Docker stack not responding to requests.

  • Return to Langfuse Cloud in your browser.
  • Open your project and click Traces in the left sidebar.
  • You should see a new trace entry. Click it to expand the span hierarchy.

Inside the trace you should see three spans corresponding to the classify, investigate, and decide nodes. Each span shows the prompt sent to the LLM, the response received, token usage, and latency.

What is a span in Langfuse?

A span represents one unit of work within a trace. In your workflow, each LangGraph node becomes its own span. The parent trace groups them together so you can see the full classify, investigate, and decide pipeline for a single dispute at a glance.

Your entire dispute pipeline is now fully observable. Next up, you will write an integration test to validate the workflow end-to-end with Pytest and enforce code quality with Ruff.

Write and Run Integration Tests

Your LangGraph workflow is now fully observable in Langfuse, with each dispute flowing through classify, investigate, and decide nodes as traced spans.

The final piece is proving that everything works together. Integration tests exercise the real endpoints, calling the actual LLM and verifying the response contract. You will also run a linter to enforce consistent code style across the entire project.

In this step, get ready to:
  • Write a Pytest integration test that validates both the health and triage endpoints.
  • Run the test suite and confirm all assertions pass.
  • Lint the codebase with Ruff to verify code quality.
Create the integration test file

FastAPI ships with a TestClient that lets you send HTTP requests to your app without starting a real server. It triggers the full lifespan (creating the database pool and Redis connection), then routes requests through your endpoints exactly as production traffic would flow.

  • Open tests/test_triage.py in your editor (the tests/ directory and its __init__.py already exist from the scaffolding step).
  • Add the imports, client setup, and health endpoint test by pasting this code:
from fastapi.testclient import TestClient

from src.main import app

client = TestClient(app)


def test_health_endpoint():
    response = client.get("/health")
    assert response.status_code == 200
    data = response.json()
    assert "status" in data

What does this code do?

TestClient(app) creates an in-process HTTP client that triggers your FastAPI lifespan (connecting to PostgreSQL and Redis) without needing a separate server process.

test_health_endpoint sends a GET to /health and asserts it returns a 200 status with a status key in the JSON body. This confirms PostgreSQL and Redis are reachable from the test environment.

  • Below the test_health_endpoint function, add the triage endpoint test:

def test_triage_endpoint():
    payload = {
        "transaction_id": "txn_test_001",
        "amount": 499.99,
        "merchant": "ElectroMart",
        "description": "I did not authorize this purchase",
        "customer_id": "cust_123",
    }
    response = client.post("/disputes/triage", json=payload)
    assert response.status_code == 200
    data = response.json()
    assert "dispute_id" in data
    assert "classification" in data
    assert data["decision"] in ["approve", "escalate", "deny"]
    assert 0.0 <= data["confidence_score"] <= 1.0

What does this test verify?

This test sends a realistic dispute payload to POST /disputes/triage. It exercises the full LangGraph workflow, calling the real LLM via LiteLLM.

  • dispute_id must be present (proves the endpoint generates a UUID).
  • classification must exist (proves the classify node ran).
  • decision must be one of approve, escalate, or deny (validates the decide node's output parsing).
  • confidence_score must fall in the 0.0 to 1.0 range (matches the Pydantic schema constraint).
  • Save tests/test_triage.py.

✔️ Awesome, I've got everything!

Great. Double check you have saved tests/test_triage.py before running the tests.

ⓧ I'd like to double check the full code

from fastapi.testclient import TestClient

from src.main import app

client = TestClient(app)


def test_health_endpoint():
    response = client.get("/health")
    assert response.status_code == 200
    data = response.json()
    assert "status" in data


def test_triage_endpoint():
    payload = {
        "transaction_id": "txn_test_001",
        "amount": 499.99,
        "merchant": "ElectroMart",
        "description": "I did not authorize this purchase",
        "customer_id": "cust_123",
    }
    response = client.post("/disputes/triage", json=payload)
    assert response.status_code == 200
    data = response.json()
    assert "dispute_id" in data
    assert "classification" in data
    assert data["decision"] in ["approve", "escalate", "deny"]
    assert 0.0 <= data["confidence_score"] <= 1.0
Run Pytest and Ruff

The tests call the real OpenAI API through the LangGraph workflow, so your OPENAI_API_KEY environment variable must be set in the terminal where you run the tests. The Docker containers must also be running so the lifespan can connect to PostgreSQL and Redis.

  • Make sure your Docker stack is still running (docker compose up from earlier).
  • Run the test suite by executing this command:
uv run pytest
  • You should see output showing 2 tests collected and both passing (marked with green dots or PASSED).

Tests failing with a connection error?

If you see an error about connecting to postgres or redis, your local machine cannot resolve the Docker Compose service hostnames. Make sure your Docker containers are running with docker compose up in a separate terminal.

If the containers are running but the hostnames still fail, add 127.0.0.1 postgres redis to your /etc/hosts file (macOS/Linux) or C:\Windows\System32\drivers\etc\hosts (Windows) so the test process can reach the containers on localhost.

Help me fix this connection error

  • Now run the linter to check code quality across the entire project:
uv run ruff check .
  • You should see no errors or warnings in the output. A clean run means your code follows the style rules defined in pyproject.toml.

What does Ruff check?

Your pyproject.toml configures Ruff to enforce rules for pyflakes errors (F), pycodestyle errors (E), import sorting (I), naming conventions (N), and warnings (W). This catches unused imports, incorrect variable naming, and style inconsistencies before they reach production.

Ruff reporting lint errors?

If Ruff flags unused imports or incorrect ordering, run uv run ruff check . --fix to auto-fix safe issues. Then re-run uv run ruff check . to confirm the output is clean.

Help me understand these lint errors

Your FinGuard AI backend is now fully tested and lint-clean. You have a working async dispute triage agent with schema validation, LLM-powered decision-making, full observability, and automated tests proving the contract holds. Next up, try the Secret Mission to add conditional routing for high-value disputes.

Secret mission

Conditional Short-Circuit Edge

Real fraud systems don't waste time investigating obvious cases. A $15,000 unauthorized charge doesn't need nuanced analysis. In this mission, you'll add a conditional edge to your LangGraph workflow that bypasses the investigation node entirely for high-value disputes, routing them straight to auto-escalation.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely on local Docker containers and makes a small number of OpenAI API calls. There are no ongoing costs once you stop sending requests to the triage endpoint.

Cost warning

Each dispute you send through the triage endpoint costs fractions of a cent in OpenAI API usage. The gpt-4o-mini pricing is $0.15 per million input tokens and $0.60 per million output tokens.

The Docker containers themselves are free since they run locally. Your total project spend is typically under $0.50.

Resources you used:

  • Docker containers (app, PostgreSQL, Redis) and the pgdata volume running on your machine.
  • Local project files in the finguard-ai folder on your Desktop.

Keep everything running

No action needed. Choose this if you are still actively building or testing new dispute payloads.

  • Your Docker containers stay running and you can keep sending disputes to http://localhost:8000/disputes/triage.
  • Each API call to the triage endpoint uses a small amount of OpenAI credit (fractions of a cent per dispute).
  • Langfuse Cloud continues to store your traces on the free tier indefinitely.

Pause - I'll come back to this later

Shut down the Docker containers to free up system memory while keeping your data and project files intact.

  • Stop all containers without removing data by running:
docker compose stop
  • When you are ready to resume, start everything back up by running:
docker compose up
  • Your PostgreSQL data in the pgdata volume is preserved between stops and starts.

Delete - I don't want to use this again

Remove all containers, volumes, and project files to start completely fresh.

  • Remove all Docker containers and the pgdata volume by running:
docker compose down -v
  • Delete the project folder from your Desktop. On macOS or Linux, run:
cd ~/Desktop
rm -rf finguard-ai
  • On Windows, run:
cd ~\Desktop
Remove-Item -Recurse -Force finguard-ai
  • Your Langfuse Cloud traces remain stored on the free tier. These have no ongoing cost.
  • OpenAI API charges have already been incurred and stop as soon as you stop making requests. No further action needed.

Nice Work!

Nice Work!

Nice work! You just built FinGuard AI, a production-grade async dispute triage backend that uses an LLM-powered agent workflow to classify, investigate, and decide on transaction disputes.

You've learned how to:

  • Build an async FastAPI backend with strict Pydantic schema contracts that orchestrates a multi-step LangGraph StateGraph workflow for AI-powered dispute triage.
  • Integrate LiteLLM for provider-agnostic LLM access and Langfuse for full pipeline observability with per-node tracing, token usage, and latency metrics.
  • Containerize the entire stack (app + PostgreSQL + Redis) with Docker Compose and validate it with Pytest integration tests and Ruff linting.
  • Secret Mission: Added conditional routing to short-circuit high-value disputes directly to escalation, bypassing the investigation node entirely.

Ready to quiz yourself?