Skip to content

About

Production-grade multi-agent orchestration system built with Python, LangGraph, FastAPI, Celery, PostgreSQL, ChromaDB, and human-in-the-loop capabilities

Resources

Stars

0 stars

Watchers

0 watching

Forks

Repository files navigation

Multi-Agent Orchestration System

Python 3.10+ FastAPI LangGraph Celery PostgreSQL 15 Redis 7 ChromaDB Streamlit Code Style: Ruff License: MIT

A resilient, production-grade enterprise multi-agent orchestration platform designed for complex, non-deterministic workflows (support ticket triaging, deep research synthesis, and autonomous operational execution).

Unlike brittle single-turn chatbots or toy agent scripts, this system delivers durable execution graphs, a tri-tier memory hierarchy, model-agnostic tool dispatching, distributed asynchronous queueing, interactive Human-in-the-Loop (HITL) checkpoints, circuit-breaker failure recovery, and end-to-end OpenTelemetry observability.


Table of Contents


Key Architectural Highlights

  • Durable State Graph with LangGraph: Explicit state transitions, cycle prevention, and deterministic state reducers (AgentState, InputState, OutputState).
  • Tri-Tier Memory:
    • Working Memory: LangGraph typed state + Redis checkpoints.
    • Episodic Memory: PostgreSQL relational schema for full audit trails of tasks, runs, messages, tool calls, and decisions.
    • Semantic Memory: ChromaDB vector store for knowledge base retrieval (RAG) and past resolution similarity matching.
  • Provider-Neutral Tool Registry: One canonical JSON Schema per tool, adapted dynamically at runtime into OpenAI Function Calling or Anthropic Tool Use payloads without provider lock-in.
  • Off-Request-Path Async Execution: FastAPI offloads heavy graph executions to Celery workers backed by Redis, streaming progress events in real-time via Redis Pub/Sub and SSE.
  • Resilience & Circuit Breakers: Automatic exponential backoff retries for transient tool/LLM errors; trips into HITL escalation on consecutive failures to prevent runaway API loops.
  • First-Class Human-in-the-Loop (HITL): Critic-driven confidence gating, interactive operator review UI in Streamlit, and zero-loss resume-from-checkpoint.
  • Production Observability: Correlation ID propagation across HTTP requests, Celery jobs, LangGraph nodes, and tool calls; OpenTelemetry spans, Prometheus metrics, and JSON structured logs.
  • Quantitative Evaluation Harness: Automated offline evaluation framework testing agent correctness, tool selection accuracy, and escalation appropriateness against benchmark golden datasets.

System Architecture & Data Flow

flowchart TB
    subgraph Clients["Client Layer"]
        UI["Streamlit Operator UI (:8501)"]
        External["External Systems / Webhooks"]
    end

    subgraph Ingress["Ingress & Storage of Record"]
        API["FastAPI Application (:8000)"]
        PG[("PostgreSQL 15<br/>(Durable System of Record)")]
    end

    subgraph AsyncQueue["Async Distributed Queue"]
        RedisQueue[("Redis 7 Broker & Cache")]
        CeleryWorker["Celery Graph Workers"]
    end

    subgraph Orchestration["Agent Orchestration Engine (LangGraph)"]
        LG["State Graph Engine"]
        Router["Router / Planner Agent"]
        Researcher["Researcher Agent"]
        Executor["Executor Agent"]
        Critic["Critic / Evaluator Agent"]
        Escalation["Escalation Agent"]
    end

    subgraph Services["Memory & Tooling Services"]
        Chroma[("ChromaDB Vector Store")]
        ToolReg["Tool Registry & Adapters"]
        ExtAPIs["External APIs & Services"]
        LLMs["OpenAI / Anthropic LLMs"]
    end

    subgraph Telemetry["Observability & Auditing"]
        OTel["OpenTelemetry Traces & Spans"]
        Prom["Prometheus Metrics"]
        Logs["Structured JSON Logs"]
    end

    UI -->|"HTTP / REST / SSE"| API
    External -->|"HTTP REST API"| API
    API -->|"Persist Task & Audit Log"| PG
    API -->|"Enqueue Job"| RedisQueue
    RedisQueue -->|"Consume Task"| CeleryWorker
    CeleryWorker -->|"Execute Graph"| LG

    LG --> Router
    LG --> Researcher
    LG --> Executor
    LG --> Critic
    LG --> Escalation

    Researcher -->|"Semantic Search (RAG)"| Chroma
    Executor -->|"Validate & Dispatch"| ToolReg
    ToolReg -->|"Canonical Adapters"| LLMs
    ToolReg -->|"Execute Integrations"| ExtAPIs

    LG -.->|"Checkpoints"| RedisQueue
    LG -.->|"State Diffs & Events"| PG
    CeleryWorker -.->|"Pub/Sub Progress"| RedisQueue
    RedisQueue -.->|"Real-Time Push"| UI

    API -.-> OTel
    CeleryWorker -.-> OTel
    LG -.-> Prom
    API -.-> Logs
Loading

Multi-Agent Collaboration Graph

stateDiagram-v2
    [*] --> Router: Task Submitted (QUEUED)

    state Router {
        [*] --> ClassifyIntent
        ClassifyIntent --> FormulatePlan
        FormulatePlan --> [*]
    }

    Router --> Researcher: Needs Retrieval / RAG
    Router --> Executor: Direct Execution Ready

    state Researcher {
        [*] --> QueryChromaDB
        QueryChromaDB --> WebSearchFallback
        WebSearchFallback --> SynthesizeContext
        SynthesizeContext --> [*]
    }

    Researcher --> Executor: Context Assembled

    state Executor {
        [*] --> ValidateToolParameters
        ValidateToolParameters --> ExecuteTools
        ExecuteTools --> FormatExecutionResult
        FormatExecutionResult --> [*]
    }

    Executor --> Critic: Verify & Score Output

    state Critic {
        [*] --> EvaluateConfidence
        EvaluateConfidence --> CheckRiskPolicies
        CheckRiskPolicies --> RouteOutcome
    }

    Critic --> [*]: High Confidence (COMPLETED)
    Critic --> Executor: Recoverable Flaw (RETRY with Feedback)
    Critic --> Escalation: Low Confidence / High Risk Action

    state Escalation {
        [*] --> FreezeCheckpoint
        FreezeCheckpoint --> CreateHITLRequest
        CreateHITLRequest --> AwaitOperatorDecision
    }

    Escalation --> Router: Human Approved / Edited (RESUMING)
    Escalation --> [*]: Human Rejected / Aborted (CANCELLED)
Loading

Agent Roles & Responsibilities

Agent Responsibilities Primary Tools / Context
Router / Planner Ingests the raw user prompt, classifies task domain, decomposes complex goals into sub-actions, and determines the graph routing trajectory. Working memory, system prompts, task metadata.
Researcher Performs semantic similarity search against indexed technical docs and knowledge bases; queries live web engines for external grounding. kb_retrieval_tool, web_search_tool, ChromaDB embeddings.
Executor Resolves required tool schemas, translates arguments to provider-specific function payloads, and executes API/database actions safely. Tool Registry, internal_api_tool, external microservice connectors.
Critic / Evaluator Validates generated answers against retrieved grounding to prevent hallucination; scores confidence (0.0–1.0) and assesses operational risk. Grounding heuristics, risk rules, self-critique rubric.
Escalation Serializes graph checkpoint state into PostgreSQL, creates a pending HITL request, halts execution, and handles resume payloads upon human resolution. hitl_requests repository, Redis checkpointer.

Tri-Tier Memory Hierarchy

To balance millisecond execution latency with multi-tenant durability and rich semantic retrieval, state is segregated into three dedicated tiers:

+-----------------------------------------------------------------------------------+
| 1. WORKING MEMORY (Ephemeral, Single-Run)                                         |
|    - LangGraph TypedState (InputState, OutputState, AgentState)                   |
|    - In-Memory Annotated reducers (append_list, merge_dict)                       |
|    - Redis checkpoint cache for mid-flight node state and circuit breaker counters |
+-----------------------------------------------------------------------------------+
                                         │
                                         ▼
+-----------------------------------------------------------------------------------+
| 2. EPISODIC MEMORY (Durable, Relational System of Record)                         |
|    - PostgreSQL 15 (SQLAlchemy 2.0 + JSONB)                                       |
|    - Full audit history: users, tasks, runs, messages, tool_calls, hitl_requests  |
|    - Comprehensive timeline replayability for debugging and legal compliance     |
+-----------------------------------------------------------------------------------+
                                         │
                                         ▼
+-----------------------------------------------------------------------------------+
| 3. SEMANTIC MEMORY (Durable, Vector-Indexed Knowledge)                            |
|    - ChromaDB persistent collections with cosine similarity                      |
|    - Product knowledge base docs, technical manuals, internal policy guidelines   |
|    - Historical resolved task index for few-shot exemplar retrieval (RAG)        |
+-----------------------------------------------------------------------------------+

Tool System & Dual-LLM Adapters

The tool subsystem ensures agents remain entirely model-agnostic. Tools are written once as standard Python callables backed by canonical JSON Schema specifications.

Canonical JSON Schema
    (src/tools/schemas/*.json)
             │
             ├──► OpenAI Adapter ────► {"type": "function", "function": {...}}
             │
             └──► Anthropic Adapter ──► {"name": ..., "input_schema": {...}}

Core Features

  • JSON Schema Validation: Strict input validation using standard JSON schemas with additionalProperties: false.
  • Zero Schema Bleed: Deepcopy isolation prevents runtime parameter mutations across concurrent requests.
  • Provider Interoperability: Switch effortlessly between OpenAI (gpt-4o) and Anthropic (claude-3-5-sonnet-20241022) dynamically without rewriting tool code.
  • Audited Execution: Every invocation records execution duration, return codes, input args, and outputs directly to the PostgreSQL tool_calls table.

Human-In-The-Loop (HITL) Lifecycle

When the Critic Agent flags a low confidence score, detects policy risk (e.g., destructive database operations or external emails), or when the circuit breaker trips after repeated tool failures, execution pauses safely:

sequenceDiagram
    autonumber
    actor User as Client / User
    participant API as FastAPI
    participant Q as Redis / Celery
    participant LG as LangGraph
    participant Critic as Critic Agent
    participant PG as PostgreSQL
    actor Operator as Human Operator (Streamlit)

    User->>API: POST /api/v1/tasks (Submit task)
    API->>PG: INSERT tasks (status: QUEUED)
    API->>Q: Enqueue Celery task
    Q->>LG: Execute Graph
    LG->>Critic: Evaluate output
    Critic->>PG: INSERT hitl_requests (status: PENDING)
    Critic->>PG: UPDATE tasks (status: PENDING_HUMAN)
    Critic->>Q: Pause Graph & Save Checkpoint
    
    Operator->>API: GET /api/v1/hitl/pending
    API-->>Operator: Return pending escalation details & diffs
    Operator->>API: POST /api/v1/hitl/{id}/decision (APPROVE / EDIT / REJECT)
    API->>PG: UPDATE hitl_requests (status: APPROVED/EDITED)
    API->>PG: UPDATE tasks (status: RESUMING)
    API->>Q: Enqueue Resume Task with human decision context
    Q->>LG: Load Checkpoint + Inject Human State
    LG->>LG: Complete remaining nodes
    LG->>PG: UPDATE tasks (status: COMPLETED)
Loading

Repository Structure

.
├── docker-compose.yml              # Multi-container stack (Postgres, Redis, Chroma, API, Worker, Streamlit)
├── Dockerfile                      # Multi-stage production container build
├── pyproject.toml                  # Python package specifications & tool configurations
├── alembic.ini                     # Alembic database migration config
├── .env.example                    # Environment variable template
├── AGENTS.md                       # Engineering governance rules for AI coding pairs
├── migrations/                     # Alembic schema version scripts
│   └── versions/                   # DDL migration scripts
├── scripts/
│   ├── init_db.sh                  # Database initialization script
│   ├── seed_kb.py                  # Knowledge-base ingestion into ChromaDB
│   └── run_eval.sh                 # Evaluation runner script
├── src/
│   ├── config.py                   # Pydantic Settings application config
│   ├── api/                        # FastAPI service layer
│   │   ├── main.py                 # FastAPI application factory & middleware
│   │   ├── deps.py                 # Dependency injection (sessions, auth, repos)
│   │   ├── routes/                 # API endpoint routers
│   │   │   ├── tasks.py            # Task dispatch, status polling, listing
│   │   │   ├── hitl.py             # Human-in-the-loop review queue & resolution
│   │   │   └── health.py           # Readiness & liveness healthchecks
│   │   └── schemas/                # Pydantic v2 request/response schemas
│   ├── agents/                     # Specialized agent implementations
│   │   ├── router_agent.py         # Intent classification & task routing
│   │   ├── researcher_agent.py     # RAG context retriever
│   │   ├── executor_agent.py       # Deterministic tool caller
│   │   ├── critic_agent.py         # Quality evaluator & confidence scorer
│   │   └── escalation_agent.py     # HITL suspension & escalation packager
│   ├── graph/                      # LangGraph state graph wiring
│   │   ├── state.py                # TypedDict state contracts & reducers
│   │   ├── build_graph.py          # Node definitions & conditional routing edges
│   │   └── checkpointer.py         # Redis-backed durable checkpoint engine
│   ├── tools/                      # Tool registry and adapters
│   │   ├── registry.py             # Core tool catalog & validator
│   │   ├── loader.py               # Dynamic schema loader
│   │   ├── openai_adapter.py       # OpenAI function calling format translator
│   │   ├── anthropic_adapter.py    # Anthropic tool use format translator
│   │   ├── kb_retrieval_tool.py    # Knowledge-base retrieval tool
│   │   ├── web_search_tool.py      # External search engine wrapper
│   │   └── schemas/                # Canonical JSON schemas per tool
│   ├── llm/                        # Unified LLM provider clients
│   │   ├── openai_client.py        # Async OpenAI client wrapper
│   │   ├── anthropic_client.py     # Async Anthropic client wrapper
│   │   └── router.py               # Model router with fallback support
│   ├── memory/                     # Memory access interfaces
│   │   ├── short_term.py           # In-memory working scratchpad
│   │   ├── episodic_store.py       # Postgres episodic conversation history
│   │   └── semantic_store.py       # ChromaDB vector embedding & RAG client
│   ├── db/                         # PostgreSQL Data Layer (SQLAlchemy 2.0)
│   │   ├── session.py              # Engine & SessionLocal factory
│   │   ├── models.py               # Declarative ORM models & enums
│   │   └── repositories/           # Repository pattern for database entities
│   │       ├── task.py             # Task entity CRUD & queries
│   │       ├── run.py              # Run execution records
│   │       ├── message.py          # Message stream repository
│   │       ├── tool_call.py        # Tool invocation audit repository
│   │       ├── hitl_request.py     # HITL escalation queue management
│   │       ├── audit_log.py        # Immutable audit logging
│   │       └── eval_result.py      # Evaluation benchmark persistence
│   ├── celery_app/                 # Async task processing
│   │   ├── celery_config.py        # Broker, backend, and serialization settings
│   │   ├── worker.py               # Celery worker process entrypoint
│   │   └── tasks.py                # Graph execution worker jobs & circuit breakers
│   ├── observability/              # Telemetry, logging & metrics
│   │   ├── logging_config.py       # JSON-structured logging with contextvars
│   │   ├── tracing.py              # OpenTelemetry tracer & span injectors
│   │   └── metrics.py              # Prometheus latency, error, and queue counters
│   └── evaluation/                 # Agent benchmarking suite
│       ├── eval_runner.py          # Batch evaluation execution harness
│       ├── scorers.py              # Correctness & safety scoring heuristics
│       └── datasets/               # Curated evaluation benchmark tasks
├── frontend/
│   └── streamlit_app/              # Operator dashboard & HITL management UI
│       ├── app.py                  # Streamlit entry point
│       └── pages/
│           ├── submit_task.py      # Interactive task submission
│           ├── task_status.py      # Live run streaming & state visualizer
│           └── hitl_queue.py       # Escalation review queue & approval portal
├── docs/                           # Architecture specs & phase roadmaps
│   ├── ARCHITECTURE.md
│   └── PHASES.md
└── tests/                          # Automated test suites
    ├── unit/                       # Fast isolated tests (models, adapters, tools)
    ├── integration/                # DB repositories, Celery tasks, graph flow
    └── e2e/                        # Full HTTP API and HITL flow validations

Quickstart Guide

Prerequisites

  • Docker & Docker Compose (Recommended for turnkey deployment)
  • Alternatively, for local manual dev:
    • Python 3.10+
    • PostgreSQL 15+
    • Redis 7+
    • ChromaDB

1. Environment Configuration

Copy the example environment configuration:

cp .env.example .env

Edit .env to configure your LLM provider API keys and database credentials:

# LLM Credentials
OPENAI_API_KEY=sk-...
OPENAI_MODEL=gpt-4o

ANTHROPIC_API_KEY=sk-ant-...
ANTHROPIC_MODEL=claude-3-5-sonnet-20241022

# System Defaults
ENVIRONMENT=development
LOG_LEVEL=INFO

2. One-Click Stack Startup (Docker Compose)

Launch the entire containerized architecture (PostgreSQL, Redis, ChromaDB, FastAPI API, Celery Worker, and Streamlit Dashboard):

docker compose up --build -d

Check the health of all running services:

docker compose ps
Service Port Description
FastAPI http://localhost:8000 OpenAPI Docs: http://localhost:8000/docs
Streamlit UI http://localhost:8501 Operator Dashboard & HITL Queue
PostgreSQL localhost:5432 System of record database
Redis localhost:6379 Queue broker & checkpoint store
ChromaDB http://localhost:8000 (internal: 8000) Vector memory service

3. Local Development Setup

If you prefer to run services natively on your host machine:

Step A: Install Dependencies & Activate Virtual Environment

python3 -m venv .venv
source .venv/bin/activate
pip install --upgrade pip
pip install -e ".[dev]"

Step B: Start Infrastructure Containers

Start PostgreSQL, Redis, and ChromaDB via Docker:

docker compose up -d postgres redis chroma

Step C: Run Database Migrations

Apply the Alembic schema migrations to instantiate tables, indexes, and constraints:

alembic upgrade head

Step D: Seed the Semantic Knowledge Base

Index sample operational runbooks, technical manuals, and policies into ChromaDB:

python scripts/seed_kb.py

Step E: Start Application Processes

In separate terminal tabs (or via process managers):

  1. Start the Celery Worker:

    celery -A src.celery_app.worker worker --loglevel=info -c 4
  2. Start the FastAPI Backend:

    uvicorn src.api.main:app --host 0.0.0.0 --port 8000 --reload
  3. Start the Streamlit Operator UI:

    streamlit run frontend/streamlit_app/app.py --server.port 8501

REST API Reference

The FastAPI service exposes an OpenAPI 3.1 specification accessible at /docs or /redoc.

1. Submit a New Task

POST /api/v1/tasks

Asynchronously ingests a task, establishes the database record, and queues the execution.

curl -X POST http://localhost:8000/api/v1/tasks \
  -H "Content-Type: application/json" \
  -d '{
    "input_text": "Investigate why customer ID 4892 encountered high latency on checkout and propose remediation.",
    "task_type": "operational_query",
    "metadata": {"priority": "high", "department": "billing"}
  }'

Response (202 Accepted):

{
  "task_id": "7b687f8f-2877-44c7-a4dc-215c0e4c6328",
  "status": "QUEUED",
  "created_at": "2026-09-21T03:30:00Z"
}

2. Poll Task Execution Status

GET /api/v1/tasks/{task_id}

curl -X GET http://localhost:8000/api/v1/tasks/7b687f8f-2877-44c7-a4dc-215c0e4c6328

Response (200 OK):

{
  "task_id": "7b687f8f-2877-44c7-a4dc-215c0e4c6328",
  "status": "COMPLETED",
  "result": {
    "summary": "Root cause identified: Database connection exhaustion in checkout microservice due to unindexed query.",
    "confidence_score": 0.94,
    "actions_taken": [
      "kb_retrieval_tool: fetched checkout architecture docs",
      "internal_api_tool: analyzed service log metrics"
    ]
  },
  "created_at": "2026-09-21T03:30:00Z",
  "completed_at": "2026-09-21T03:30:14Z"
}

3. Review Pending HITL Escalations

GET /api/v1/hitl/pending

Returns tasks currently paused in PENDING_HUMAN awaiting operator judgment.

curl -X GET http://localhost:8000/api/v1/hitl/pending

Response (200 OK):

[
  {
    "hitl_request_id": "11a8b92d-90c1-4235-94be-df92c30084ba",
    "task_id": "7b687f8f-2877-44c7-a4dc-215c0e4c6328",
    "status": "PENDING",
    "reason": "Critic flagged high-risk operation: Proposed tool invocation modifies database connection pool configuration.",
    "proposed_action": {
      "tool": "internal_api_tool",
      "parameters": {"action": "restart_pool", "pool_size": 50}
    },
    "created_at": "2026-09-21T03:30:08Z"
  }
]

4. Submit HITL Resolution

POST /api/v1/hitl/{hitl_request_id}/decision

curl -X POST http://localhost:8000/api/v1/hitl/11a8b92d-90c1-4235-94be-df92c30084ba/decision \
  -H "Content-Type: application/json" \
  -d '{
    "decision": "APPROVED",
    "operator_notes": "Verified connection pool capacity with DevOps team. Safe to proceed.",
    "override_parameters": null
  }'

Response (200 OK):

{
  "hitl_request_id": "11a8b92d-90c1-4235-94be-df92c30084ba",
  "status": "APPROVED",
  "task_status": "RESUMING",
  "resumed_at": "2026-09-21T03:32:01Z"
}

Streamlit Operator Dashboard

Access the interactive dashboard at http://localhost:8501.

┌────────────────────────────────────────────────────────────────────────┐
│  Multi-Agent Orchestrator Operator Portal                              │
├─────────────────┬──────────────────────────────────────────────────────┤
│ 📋 Submit Task   │ ▶ Live Graph Execution Trace                         │
│                 │                                                      │
│ 📊 Task Monitor │ [Router]  ──► Intent: Operational Remediation         │
│                 │ [Search]  ──► RAG: Retrieved 3 runbook snippets      │
│ ⚠️ HITL Queue   │ [Execute] ──► Call: internal_api_tool (read_logs)    │
│    (1 Pending)  │ [Critic]  ──► Confidence: 0.94 | Risk: Low           │
│                 │ Status: COMPLETED in 4.2s                            │
└─────────────────┴──────────────────────────────────────────────────────┘
  • Submit Task (pages/submit_task.py): Interface for dispatching ad-hoc or batch tasks with custom metadata, priority flags, and model selection.
  • Task Monitor (pages/task_status.py): Real-time state inspection displaying active LangGraph nodes, tool inputs/outputs, and token consumption metrics via Redis pub/sub.
  • HITL Review Queue (pages/hitl_queue.py): Visual escalation inbox allowing human operators to inspect agent reasoning, inspect tool payloads, edit parameters, or approve/abort actions with one click.

Observability, Telemetry & Tracing

The platform incorporates comprehensive instrumentation for full-lifecycle observability:

1. Correlation & Context Tracking

Every incoming request receives an X-Correlation-ID header. Python contextvars propagate this identifier across async boundaries, Celery task metadata, LangGraph state, and tool audit logs.

2. OpenTelemetry Tracing

Distributed tracing tracks spans end-to-end:

[HTTP POST /tasks]
  └── [Celery Task: run_graph_task]
        └── [LangGraph: RouterAgent]
        └── [LangGraph: ResearcherAgent]
              └── [ChromaDB: QueryEmbeddings]
        └── [LangGraph: ExecutorAgent]
              └── [Tool: kb_retrieval_tool]
        └── [LangGraph: CriticAgent]

3. Prometheus Metrics (GET /metrics)

  • agent_task_duration_seconds: Histogram of end-to-end task completion latency.
  • agent_node_executions_total: Counter of agent node transitions partitioned by agent name.
  • agent_tool_calls_total: Counter tracking tool invocations labelled by tool name and status (SUCCESS, TIMEOUT, FAILURE).
  • agent_hitl_escalations_total: Counter of triggered human-in-the-loop escalations.
  • agent_circuit_breaker_trips_total: Counter of circuit breaker trips triggered by consecutive errors.

Evaluation Harness & Quality Assurance

To prevent prompt regressions and validate new agent graph topologies, the offline evaluation harness measures performance against a curated dataset:

# Execute the offline benchmark suite
./scripts/run_eval.sh

Evaluation Dimensions

  1. Semantic Correctness: Compares final synthesized answers against golden reference outputs using LLM-as-a-judge scoring (0.0 to 1.0).
  2. Tool-Call Precision & Recall: Verifies that the correct tools were selected and invoked with schema-compliant arguments.
  3. Escalation Appropriateness: Validates that risky or ambiguous synthetic tasks accurately trigger PENDING_HUMAN escalations instead of guessing or hallucinating.

Results are persisted in the eval_results database table and output as markdown comparison reports in CI/CD pipelines.


Testing & Code Quality

The system is tested using a multi-layered verification strategy:

# 1. Run Unit Tests (Fast, In-Memory Models & Adapters)
pytest tests/unit

# 2. Run Integration Tests (PostgreSQL Repositories & Graph Transitions)
pytest tests/integration

# 3. Run End-to-End Tests (Full HTTP API & HITL Escalation Cycles)
pytest tests/e2e

# 4. Run Complete Suite with Coverage
pytest --cov=src tests/

# 5. Run Static Code Quality & Formatting Checks
ruff check src tests
ruff format --check src tests

Configuration Reference

All settings can be configured via environment variables or a local .env file:

Variable Type Default Description
ENVIRONMENT string development Deployment environment (development, staging, production)
DEBUG boolean false Enable verbose logging and hot-reload
LOG_LEVEL string INFO Logging level (DEBUG, INFO, WARNING, ERROR)
API_HOST string 0.0.0.0 Bind host for the FastAPI application
API_PORT integer 8000 Port for the FastAPI application
STREAMLIT_PORT integer 8501 Port for the operator Streamlit dashboard
POSTGRES_HOST string localhost PostgreSQL hostname
POSTGRES_PORT integer 5432 PostgreSQL port
POSTGRES_USER string postgres PostgreSQL username
POSTGRES_PASSWORD string postgres PostgreSQL password
POSTGRES_DB string orchestrator PostgreSQL database name
DATABASE_URL string (derived) Explicit database connection override
REDIS_HOST string localhost Redis hostname
REDIS_PORT integer 6379 Redis port
REDIS_DB integer 0 Redis database index
CELERY_BROKER_URL string (derived) Override for Celery broker URL
CHROMA_HOST string localhost ChromaDB vector store host
CHROMA_PORT integer 8000 ChromaDB vector store port
OPENAI_API_KEY string None OpenAI API Secret Key
OPENAI_MODEL string gpt-4o Default OpenAI model name
ANTHROPIC_API_KEY string None Anthropic API Secret Key
ANTHROPIC_MODEL string claude-3-5-sonnet-20241022 Default Anthropic model name

Production Hardening & Deployment

For enterprise production deployments, the following architectural safeguards are built-in:

  1. Multi-Stage Non-Root Docker Image: The Dockerfile utilizes a multi-stage build removing build toolchains from the runtime image and executes under an unprivileged appuser (uid=10001).
  2. Graceful Degradation & Backpressure: Celery task concurrency limits prevent API key rate quota exhaustion; queue depth alerts inform autoscaling rules.
  3. Audit Trail Immutability: All state changes, human intervention decisions, and tool executions write write-once log rows to PostgreSQL with UTC timestamps.
  4. Secret Sanitization: Logging formatters automatically redact authorization tokens, API keys, and sensitive payload strings matching PII patterns.

License

This project is licensed under the terms of the MIT License.

About

Production-grade multi-agent orchestration system built with Python, LangGraph, FastAPI, Celery, PostgreSQL, ChromaDB, and human-in-the-loop capabilities

Resources

Stars

0 stars

Watchers

0 watching

Forks

Releases

Packages

Contributors

Languages