FDE AI Platform

📦 Archived — superseded by Forgehand.

This repository is the minimal, didactic version of the idea: agent workflows as explicit, typed steps behind a stable API, with an evaluation verdict and trace metadata on every run. Its full evolution — LangGraph orchestration, parallel fan-out, an LLM judge with objective veto, human gates, budget circuit breakers and durable execution — lives in Forgehand. The code here stays readable as a compact reference, but all active development happens there.

A workflow orchestration platform for LLM agents: define agent workflows as explicit, typed steps; run them behind a stable FastAPI contract; and get an evaluation verdict plus full trace metadata back with every single run.

The Problem

Most teams ship their first LLM agent as a script or a notebook: one prompt, one loop, no contract. That works until someone asks the operational questions — Which steps ran for this request? Which tools were called, with what inputs? Why was this answer escalated? How long did the run take? Did this response go through an LLM at all?

Loose scripts cannot answer these questions because they have no run identity, no step log, no evaluation gate, and no API boundary. A platform for LLM agent workflows makes the workflow itself a first-class, observable object: every run is traced, scored, and returned through a versioned API contract that clients can build against.

The Solution

FDE AI Platform is a small, production-shaped foundation for exactly that:

• Workflow orchestration — an AgentWorkflow that executes a fixed pipeline of explicit steps (policy_check → intent_analysis → tool_routing → response_composition) over a shared, typed WorkflowState.
• Typed API contracts — FastAPI + Pydantic models (RunRequest / RunResponse) so the workflow is consumed as a service, not imported as a script.
• Evaluation hooks — every run passes through a RunEvaluator that scores deterministic checks (answer present, intent set, policy compliance, escalation handling, tool routing) and attaches the verdict to the response.
• Observability metadata — each run carries a run_id, trace_id, latency, executed steps, tool calls, policy flags, and the active generation mode.
• Pluggable answer generation — the final answer comes from a generator behind a small interface: Claude-backed when a key is configured, a deterministic template otherwise, so tests and CI never need secrets.
• Operational packaging — Dockerfile, Docker Compose, Makefile, and a secrets-free GitHub Actions CI pipeline.

Architecture

Client
  |
  v
FastAPI  (POST /runs — typed RunRequest/RunResponse contract)
  |
  v
AgentWorkflow (orchestrator)
  |-- policy_check          -> policy flags (PII markers, high priority)
  |-- intent_analysis       -> Intent: support | analysis | escalation | general
  |-- tool_routing          -> allowlisted ToolRegistry calls per intent
  |-- response_composition  -> AnswerGenerator (Claude or deterministic)
  |
  v
RunEvaluator (per-run checks + score)
  |
  v
RunResponse = answer + evaluation + RunMetadata (run_id, trace_id, latency,
              steps, tool calls, policy flags, generation mode)

See docs/architecture.md for the request flow and the production boundary (persistence and async execution are intentionally kept out of the hot path today).

Quickstart

git clone https://github.com/lucianoon/fde-ai-platform.git
cd fde-ai-platform
cp .env.example .env

make install   # venv + dependencies
make test      # pytest, no API keys required
make dev       # uvicorn on http://localhost:8000

Or with Docker Compose (starts the API plus PostgreSQL and Redis containers, which are provisioned for the planned persistence/queue adapters):

make docker-up

Then exercise the API:

curl -X POST http://localhost:8000/runs \
  -H "Content-Type: application/json" \
  -d '{
    "tenant_id": "acme",
    "user_message": "Analyze this customer request and recommend the next action.",
    "context": {"channel": "support", "priority": "high"}
  }'

Defining a Workflow

A workflow is a class that runs explicit steps over a typed WorkflowState and records everything it does. The built-in AgentWorkflow is the reference implementation:

from fde_ai_platform.models import RunRequest
from fde_ai_platform.workflow import AgentWorkflow

workflow = AgentWorkflow()          # tools, evaluator and generator are injectable
response = workflow.run(
    RunRequest(
        tenant_id="acme",
        user_message="Customer needs support with a billing ticket.",
        context={"priority": "normal"},
    )
)

response.intent                      # Intent.SUPPORT
response.metadata.steps              # ["policy_check", "intent_analysis", ...]
response.metadata.tool_calls         # [ToolCall(name="ticket_triage", ...)]
response.evaluation.passed           # True

Its constructor takes three collaborators, each replaceable without touching the API layer:

Collaborator	Contract	Default
tools	ToolRegistry.run(name, payload) -> ToolCall — allowlisted; unknown tools raise	knowledge_lookup, ticket_triage, metrics_lookup
evaluator	RunEvaluator.evaluate(state) -> EvaluationResult	five deterministic checks (see below)
answer_generator	AnswerGenerator.compose(state) -> str	picked from the environment (see below)

Routing rules are deterministic: intent is classified from the message, support routes to ticket_triage, analysis routes to metrics_lookup + knowledge_lookup, and high-priority or escalation-intent runs are flagged requires_escalation. Requests containing PII markers (cpf, ssn, credit card) are short-circuited with a deterministic refusal and are never sent to an LLM.

Answer generation modes

The last step delegates to a pluggable generator selected by FDE_LLM_MODE:

• deterministic — intent-based template answer; zero network access.
• llm — Claude (Anthropic SDK) synthesizes the answer grounded only in the classified intent and the collected tool outputs; model set via FDE_LLM_MODEL.
• auto (default) — llm when ANTHROPIC_API_KEY is set, otherwise deterministic.

A transient API error falls back to the deterministic template, and every response reports which mode actually produced the answer via metadata.generation_mode.

Observability

Every run returns a RunMetadata block designed to answer operational questions without log spelunking:

{
  "run_id": "0b2e…",
  "trace_id": "9f41…",
  "latency_ms": 1.8,
  "steps": ["policy_check", "intent_analysis", "tool_routing", "response_composition"],
  "tool_calls": [{"name": "ticket_triage", "input": {"priority": "high"}, "output": {"queue": "human-review", "priority": "high"}}],
  "policy_flags": ["high_priority"],
  "generation_mode": "deterministic"
}

• run_id / trace_id give each execution a stable identity for correlation.
• steps is the ordered log of workflow stages that actually ran.
• tool_calls records every tool invocation with its input and output.
• policy_flags surfaces guardrail hits (pii_detected, high_priority).
• latency_ms is measured per run by the RunTrace primitive.
• generation_mode makes it auditable whether an LLM produced the answer.

Evaluation Hooks

Evaluation is part of every response, not an offline afterthought. The RunEvaluator runs five deterministic checks and returns a verdict with the run:

Check	Passes when
has_answer	the composed answer is non-empty
has_intent	an intent was classified
policy_ok	no PII marker was detected
escalation_ok	high-priority runs were flagged for escalation
tool_routing_ok	support/analysis intents actually called tools

The result (passed, score = fraction of checks passed, per-check booleans, human-readable notes) ships inside RunResponse.evaluation, so clients and dashboards can gate on quality per request. The evaluator is injectable, so stricter or LLM-assisted checks can be swapped in behind the same interface.

API

Method	Path	Description
GET	/health	Liveness check, returns {"status": "ok"}
POST	/runs	Execute a workflow run; body is RunRequest, response is RunResponse

Contracts are Pydantic models in src/fde_ai_platform/models.py; interactive docs are served at /docs when the API is running.

Testing

make test
# or
PYTHONPATH=src pytest -q

The suite covers the API contract (tests/test_api.py), workflow routing/escalation/PII behavior (tests/test_workflow.py), and generator selection with no network access (tests/test_generation.py). Tests require no API keys — generation resolves to the deterministic mode — and the same secrets-free suite runs in CI on every push and pull request.

Roadmap

• PostgreSQL run persistence
• Redis-backed job queue for async runs
• OpenTelemetry exporter for run traces
• Prometheus metrics endpoint
• Authentication and tenant isolation

License

MIT — Copyright (c) 2026 Luciano de Oliveira Nunes.