Backend & API
The FastAPI + LangGraph + Celery backbone of the FCA support agent. Each sub-page covers one architectural pattern verbatim from full_project_context_updated.txt:
- Specialist Agent Deep Dives & LangGraph Flow — intent classification, specialist routing, compliance gating, SSE streaming
- FastAPI SSE Streaming — per-request
ContextVarqueue +SSELogHandler - LangGraph Multi-Agent Streaming — node-by-node
stream_messagetuples - LangGraph Checkpointing (AsyncPostgresSaver) — deadlock-safe DDL on first init, graph persist across restarts
- LangGraph Human-in-the-Loop — interrupt graph execution for manual approval
- Multi-Agent Supervisor Routing —
IntentClassifier+ conditional edge dispatch - Async Repository Pattern — generic
AsyncRepository[ModelT]withcreate,get,filter - Async Service Layer (Context Manager) — async context manager for session lifecycle
- FastAPI Lifespan Startup/Shutdown —
@asynccontextmanagerlifespan forinit_db/close_db - Structured Logging with JSONFormatter —
ContextVar[asyncio.Queue]for live SSE log piping - FastAPI Depends() for Auth/RBAC — scope-checking
Depends()guard on every protected route - Langfuse LLM Tracing (@observe) —
@observe()decorator for end-to-end LLM call observability - Redis Cache Service & Normalized Keys —
CacheServicewith TTL, normalized key format, optional circuit breaker - Celery Async Worker Bridge — sync Celery wrapping an asyncio coroutine
- Pydantic-Settings Validation — typed
BaseSettingswith validators and computed properties - BaseAgent (Circuit Breaker + Tenacity) — fail-fast + exponential backoff for every specialist agent
- HumanAgent Escalation Priority — priority-queue escalation with
EscalationService - ComplianceChecker Hybrid Short-Circuit — redact / block / warn triage before any agent response
- WorkflowState Pydantic Schema — typed
WorkflowStatewithTypedDictfor LangGraph node I/O - Message API Route (IDOR + Scopes) — resource-level IDOR check +
read:messages/write:messagesscopes - Streamlit Chat UI (Session State + SSE) —
st.session_statemessage history +requests.get(stream=True)consumption - Per-Resource Repository Specialization — domain-specific
CustomerRepository/AccountRepositoryextendingAsyncRepository - JWT Login Route with Scopes —
POST /auth/loginreturning access token with embedded scope claims - SecurityService Sanitization Pipeline — PII redaction → Lakera injection check → allow/block triage
Last updated on