Skip to content

About

No description, website, or topics provided.

Resources

Stars

0 stars

Watchers

0 watching

Forks

Repository files navigation

🛡️ NovaTech Support Desk — LangGraph Multi-Step Enterprise Agent

Python 3.11+ LangGraph ChromaDB FastAPI Tests Passing Docker Ready License: MIT

2026 AI Engineer (AIE) & Forward-Deployed Engineer (FDE) Portfolio Core
"Build a customer support agent with tool calling, memory, and graceful failure handling."
An autonomous, stateful agentic system built with LangGraph StateGraph, ChromaDB episodic memory, dynamic working memory planning, runaway loop protection, self-healing failure recovery, and Human-In-The-Loop (HITL) checkpoints.


📌 Executive Summary & Motivation

Most LLM applications in industry fail because they rely on stateless chains or unconstrained ReAct loops. In mission-critical customer operations, these failure modes manifest as:

  1. Hallucinatory execution cycles: The agent forgets what step it is currently executing and loops indefinitely.
  2. Cascading failures: A single SQL syntax error or API timeout crashes the worker process.
  3. Unchecked irreversible side-effects: Sending customer refund emails or modifying records without human operator verification.
  4. Context amnesia: Inability to recall precedent agreements or customer concessions made in previous interactions.

This project implements a fault-tolerant, stateful agentic architecture designed to solve these challenges using LangGraph StateGraph, typed graph states, and a three-tier cognitive memory hierarchy.


🏛️ System Architecture

User Inquiry 
     │
     ▼
[ START Node ]
     │
     ▼
[ memory_recall_node ] ◄── Semantic Search in ChromaDB (Past Customer Precedents)
     │
     ▼
┌──────────────────┐
│   planner_node   │ ◄── Updates Working Memory (Dynamic Multi-Step Plan & Checklist)
└────────┬─────────┘
         │
         ▼
  [ router_node ] ── (Iteration >= Max?) ──────────────► [ abort_node (Loop Guard) ]
         │                                                            │
         ├─ (Sensitive Action: send_email?) ──► [ human_approval_node ]│
         │                                              │ (Pause / Yield)
         │                                              ▼             │
         ├─ (Tool Call?) ─────────────────────► [ tool_execution_node ]│
         │                                              │             │
         │                                       (Tool Error?)        │
         │                                              ▼             │
         │                                      [ recovery_node ]     │
         │                                              │ (Self-Healing)
         │                                              ▼             │
         │                                      (Return to Planner)   │
         ▼                                                            │
[ memory_consolidation_node ] ◄───────────────────────────────────────┘
         │ (Persist Session Summary into ChromaDB)
         ▼
    [ END Node ] ──► Response Stream & LangSmith Observability Trace

Mermaid State Diagram

graph TD
    Start([START]) --> Recall[memory_recall_node<br><i>ChromaDB Semantic Query</i>]
    Recall --> Planner[planner_node<br><i>Working Memory & Decisions</i>]
    
    Planner --> Router{router_condition}
    
    Router -- "Iteration >= Max (8)" --> Abort[abort_node<br><i>Runaway Loop Guard</i>]
    Router -- "Sensitive Action (send_email)" --> HITL[human_approval_node<br><i>HITL Checkpoint</i>]
    Router -- "Tool Request" --> Tools[tool_execution_node<br><i>Safe Dispatch</i>]
    Router -- "Finished" --> Consolidate[memory_consolidation_node<br><i>Persist to ChromaDB</i>]
    
    Tools --> ErrorCheck{tools_condition}
    ErrorCheck -- "Success" --> Planner
    ErrorCheck -- "Exception Caught" --> Recovery[recovery_node<br><i>Self-Healing Diagnosis</i>]
    Recovery --> Planner
    
    HITL -- "Awaiting Verification" --> Pause([Pause Execution & Checkpoint State])
    HITL -- "Approved / Edited / Rejected" --> Planner
    
    Abort --> Consolidate
    Consolidate --> EndNode([END])
Loading

🎯 Core Technical Features

1. Stateful Multi-Step Graph (Not Stateless Chains)

  • Typed Graph State (AgentState): Built on typing_extensions.TypedDict with LangGraph's add_messages reducer to safely accumulate message histories.
  • Resumable State Checkpoints: Graph states are persisted per thread_id using LangGraph's MemorySaver (or SQLite/Postgres checkpointers in production), enabling long-running execution and asynchronous human approvals.
  • Complete Audit Trail: Every node hop records { from_node, to_node, action_type, timestamp, details } into state["state_transitions"].

2. Three-Tier Memory Hierarchy

Modeled after human cognitive architectures:

Memory Tier Storage Mechanism Scope & Purpose
Short-Term Memory add_messages reducer + MemorySaver Preserves conversational context across multi-turn exchanges within the current thread_id.
Working Memory Typed WorkingMemory in AgentState Explicit, dynamic multi-step plan (plan), current active step index (current_step_index), and completed step log with latency (completed_steps).
Episodic Memory ChromaDB Vector Store Long-term memory across distinct sessions. Recalls past interactions for the customer (e.g. Alice's past courtesy credit for a damaged box) using semantic embeddings.

3. Production Tool Suite

  • 🗄️ database_query: Safe read-only SQL query execution over SQLite customers and orders tables. Guarded against destructive DDL/DML verbs (DROP, DELETE, UPDATE, INSERT). Includes database_schema introspection tool.
  • 🧮 calculator: Safe mathematical evaluation using Python's AST parser (calculates restocking fees, percentage deductions, and net refunds without unsafe eval()).
  • 🔍 web_search: Searches company return policies, restocking fee guidelines, and courier logistics tracking (TRK-77192-US).
  • 📝 email_draft: Composes customer support notification drafts.
  • ✉️ send_email: Irreversible sensitive action intercepted by the Human-In-The-Loop checkpoint.

4. Graceful Failure Recovery & Self-Healing

  • When a tool fails (e.g. invalid SQL table or column), tool_execution_node catches the exception without crashing the backend process.
  • The router redirects execution to recovery_node.
  • The recovery node inspects the error trace, increments error_retry_count, injects diagnostic advice (such as invoking database_schema), or falls back to an alternate tool.

5. Runaway Loop Detection Safeguard

  • Hard guardrail threshold (MAX_ITERATIONS = 8).
  • If an agent gets caught in a cyclic reasoning trap, router_condition routes to abort_node.
  • The graph terminates gracefully, summarizes completed progress from working memory, and emits an escalation ticket (LOOP-GUARD-8-8).

6. Human-In-The-Loop (HITL) Checkpoints

  • Irreversible outbound actions (send_email) are intercepted before execution.
  • The graph yields execution in a pending state and checkpoints the state.
  • Operators review the draft in the interactive dashboard, with one-click [Approve & Dispatch], [Edit Draft] (with inline modal), or [Reject] actions via /api/approve.

💻 Interactive Web Console

The project includes a clean, human-crafted web console (modeled after Linear, Supabase, and Stripe dashboards):

  • Live StateGraph Flow: Visualizes active nodes as they execute.
  • Working Memory Scratchpad: Real-time display of the current plan and completed steps with measured latency (ms) and localized timestamps.
  • State Transition Audit Trail: Chronological log of every graph transition.
  • ChromaDB Vector Browser: Live search and inspection of past session episodes.
  • SQLite Database Viewer: Table inspector for customers and orders.
  • Edit Draft Modal Dialog: Full form to edit recipient, subject, body, and audit notes before dispatch.

🚀 Quick Start Guide

Prerequisites

  • Python 3.11+
  • uv (recommended) or pip

1. Clone & Setup

git clone https://github.com/your-username/langgraph-multistep-agent.git
cd langgraph-multistep-agent

2. Create Virtual Environment & Install Dependencies

Using uv (fastest):

uv venv .venv
uv pip install -r requirements.txt --python .venv/Scripts/python.exe

Or using standard pip:

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

3. Configure Environment Variables (Optional)

cp .env.example .env

Zero-Cost / Offline Execution Mode:
If no OPENAI_API_KEY is provided, the agent runs in Deterministic Simulation Mode. All tools, multi-step planning, ChromaDB episodic recall, and HITL approvals run out of the box with zero setup cost or external API keys! When keys are provided, it automatically switches to production models (gpt-4o-mini or claude-3-5-sonnet).

4. Launch Application

python run.py

🧪 Terminal Dry Run & Test Suite

Run Interactive Terminal Dry Run

Step through every state transition directly in your terminal:

python dry_run.py

Run Automated Pytest Suite (14 Tests)

pytest -v
tests/test_episodic_memory.py::test_episodic_memory_seed_and_recall PASSED       [  7%]
tests/test_episodic_memory.py::test_episodic_memory_store_new_episode PASSED     [ 14%]
tests/test_failure_recovery.py::test_tools_condition_detects_error PASSED       [ 21%]
tests/test_failure_recovery.py::test_tool_execution_catches_exception PASSED     [ 28%]
tests/test_failure_recovery.py::test_recovery_node_suggests_alternative PASSED   [ 35%]
tests/test_hitl.py::test_router_identifies_sensitive_action PASSED              [ 42%]
tests/test_hitl.py::test_human_approval_node_pauses_for_pending PASSED          [ 50%]
tests/test_hitl.py::test_human_approval_node_executes_on_approval PASSED        [ 57%]
tests/test_hitl.py::test_human_approval_node_halts_on_rejection PASSED          [ 64%]
tests/test_loop_guard.py::test_router_detects_runaway_loop PASSED               [ 71%]
tests/test_loop_guard.py::test_abort_node_generates_safe_response PASSED        [ 78%]
tests/test_loop_guard.py::test_graph_terminates_at_max_iterations PASSED        [ 85%]
tests/test_state_machine.py::test_graph_compilation PASSED                      [ 92%]
tests/test_state_machine.py::test_working_memory_and_transitions PASSED         [100%]
============================= 14 passed in 5.58s ==============================

🐳 Docker Deployment

Run the complete stack with Docker Compose:

docker compose up --build

Access the console at http://localhost:8000.


📡 REST API Reference

Method Endpoint Description
POST /api/chat Send customer inquiry; executes multi-step graph
POST /api/approve Resume paused thread from HITL checkpoint (approved, rejected, edited)
GET /api/history/{thread_id} Retrieve state snapshot, working memory, and transition audit log
GET /api/memories/{customer_id} Inspect stored ChromaDB episodic memories
GET /api/database/customers Query SQLite customer profiles
GET /api/database/orders Query SQLite customer orders
POST /api/seed Re-seed SQLite database & ChromaDB vector store
GET /api/health Healthcheck and configuration status

🔍 Observability with LangSmith

To enable production tracing:

LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=your_langsmith_api_key_here
LANGCHAIN_PROJECT=langgraph-support-agent

Every invocation streams step-by-step latency, token counts, tool calls, and node outputs directly to your LangSmith project dashboard.


📁 Repository Structure

├── app/
│   ├── config.py              # Environment configuration & safety caps
│   ├── state.py               # TypedDict AgentState, WorkingMemory, StateTransitionRecord
│   ├── graph.py               # LangGraph StateGraph builder, conditional edges, checkpointer
│   ├── llm_engine.py          # Multi-provider LLM engine (OpenAI, Anthropic, Mock)
│   ├── main.py                # FastAPI REST API & dashboard serving
│   ├── nodes/
│   │   ├── memory_node.py     # ChromaDB episodic recall & memory consolidation
│   │   ├── planner.py         # Working memory planning & tool decision node
│   │   ├── tools_node.py      # Safe tool executor with latency & exception wrapping
│   │   ├── recovery.py        # Self-healing failure recovery node
│   │   ├── human_approval.py  # Human-in-the-loop checkpoint node
│   │   └── abort_node.py      # Runaway loop detection & graceful exit
│   ├── tools/
│   │   ├── database.py        # SQLite query & schema introspection tools
│   │   ├── calculator.py      # AST-based safe math evaluator
│   │   ├── web_search.py      # Policy lookup & courier tracking
│   │   └── email.py           # Email draft & sensitive send_email tools
│   ├── memory/
│   │   └── episodic.py        # ChromaDB vector store manager
│   ├── database/
│   │   └── seed_db.py         # SQLite schema & customer orders seed data
│   └── static/
│       ├── index.html         # Human-crafted web console
│       ├── styles.css         # Clean enterprise dashboard styling
│       └── app.js             # Client state management & modal controls
├── tests/                     # 14/14 Pytest unit test suite
├── Dockerfile                 # Container image
├── docker-compose.yml         # Container compose definition
├── requirements.txt           # Verified dependencies
├── dry_run.py                 # Terminal step-by-step walkthrough
├── run.py                     # Single-command launcher
├── PREPARATION.md             # Comprehensive interview & cross-question defense guide
└── README.md

📄 License

MIT License. Crafted for 2026 AI Engineering and Forward-Deployed Engineering interviews.

About

No description, website, or topics provided.

Resources

Stars

0 stars

Watchers

0 watching

Forks

Releases

Packages

Contributors

Languages