Diagent

Self-hosted observability for AI agents and RAG systems.

Diagent captures agent runs, spans, tool calls, and retrieval evidence through a small HTTP API and Python tracer. Finished runs are analyzed asynchronously for deterministic anomalies and, when eligible, RAG quality. Ambiguous low-quality RAG runs can then be passed to a read-only diagnostician that reasons only over persisted evidence.

Diagent is a general-purpose observability and evaluation system. Each team decides what should be observed, measured, and verified in its own application, defines the application-specific rules and expectations that matter for that system, and integrates Diagent accordingly.

To learn how to integrate your system with Diagent, see the Diagent Integration Guide .

Diagent is designed as an observability backend, not an agent framework. Your application keeps ownership of routing, tools, retrieval, generation, and domain rules; Diagent records what happened and analyzes the evidence you choose to send.

Status: early-stage project focused on a small, explainable core. The repository intentionally avoids a large tracing platform, automatic remediation, and application-specific policy logic in main.

A successful HTTP response does not tell you whether an AI workflow behaved correctly.

A run can still:

• call the same tool repeatedly,
• fail tools at an unusual rate,
• exceed expected latency or cost,
• retrieve stale or empty context,
• return a fluent answer that is weakly grounded in retrieval evidence.

Diagent makes those failures inspectable without requiring the application to move its orchestration logic into a new framework.

flowchart LR
 APP["Agent / RAG application"] -->|"Python tracer or REST"| API["FastAPI API"]
 API --> DB[("PostgreSQL")]
 API -->|"finished run"| ELIG{"RAG evaluation eligible?"}
 ELIG -->|"no"| ANOM["Rule-based anomaly detection"]
 ELIG -->|"yes"| CHAIN["Celery chain"]
 CHAIN --> ANOM2["Anomaly detection"]
 ANOM2 --> RAG["RAG quality evaluation"]
 RAG --> TRIGGER{"Diagnosis trigger eligible?"}
 TRIGGER -->|"yes"| DIAG["Read-only diagnostician"]
 TRIGGER -->|"no"| DONE["Persisted evaluation"]
 ANOM --> DB
 ANOM2 --> DB
 RAG --> DB
 DIAG --> DB
 DONE --> DB
 REDIS[("Redis")] -. "Celery broker" .-> CHAIN
 JUDGE["OpenAI or Ollama judge"] -.-> RAG
 JUDGE -.-> DIAG

The API persists telemetry first. Background analysis is handled by Celery workers through Redis. PostgreSQL remains the source of persisted runs, telemetry, alerts, evaluations, and diagnoses.

git clone https://github.com/fatihaybsn/Diagent.git
cd Diagent
cp .env.example .env
docker compose up --build -d

Check liveness:

curl http://localhost:8000/healthz

Expected response:

{"status":"ok","version":"0.1.0"}

Interactive API documentation is available at:

http://localhost:8000/docs

Telemetry ingestion and rule-based anomaly detection do not require an external judge. RAG evaluation and diagnosis do.

The default backend is OpenAI:

DIAGENT_JUDGE_BACKEND=openai
OPENAI_API_KEY=...
OPENAI_JUDGE_MODEL=gpt-4o-mini

Or use Ollama:

DIAGENT_JUDGE_BACKEND=ollama
OLLAMA_BASE_URL=http://host.docker.internal:11434
OLLAMA_JUDGE_MODEL=llama3.1

See .env.example for the full configuration surface.

Install the package from the repository:

python -m pip install -e .

Point the tracer at the running API:

export DIAGENT_API_URL=http://localhost:8000

Then instrument a function:

import diagent
@diagent.observe(agent_name="support-bot")
def answer_customer(question: str) -> str:
 diagent.log_retrieval(
 query=question,
 retrieved_chunks=[
 {
 "text": "Refund requests are accepted within 14 days.",
 "source": "refund-policy.md",
 "score": 0.91,
 }
 ],
 top_k=3,
 source_age_hours=2,
 )
 diagent.log_tool_call(
 tool_name="refund_lookup",
 args={"question": question},
 status="success",
 duration_ms=180,
 )
 return "Refund requests are accepted within 14 days."

@diagent.observe creates the run, exposes the active run context to helper functions, records success or failure, and finishes the run automatically.

For known token or cost metadata, record only values your application can actually measure:

diagent.set_run_metadata(total_tokens=420, cost_usd=0.0031)

Do not invent 0 for unknown cost or token usage.

Choose the smallest integration surface that fits your application.

Best for a Python function that represents one complete agent run.

@diagent.observe(agent_name="assistant")
def run_agent(user_input: str) -> str:
 ...

Best when your application already owns the lifecycle and needs explicit control.

from diagent.core.tracer import DiagentTracer
tracer = DiagentTracer("http://localhost:8000")
run_id = tracer.create_run(
 agent_name="assistant",
 input_text="Where is my order?",
)
tracer.log_tool_call(
 run_id,
 tool_name="order_lookup",
 args={"order_id": "A-123"},
 status="success",
 duration_ms=95,
)
tracer.finish_run(
 run_id,
 output="Your order is in transit.",
 status="finished",
)

Best for non-Python services or custom integration layers.

curl -X POST http://localhost:8000/runs \
 -H "Content-Type: application/json" \
 -d '{"agent_name":"support-bot","input":"Where is my order?"}'

Then attach telemetry to the returned run ID:

POST /runs/{run_id}/spans
POST /runs/{run_id}/tool_calls
POST /runs/{run_id}/retrievals
POST /runs/{run_id}/finish

For the complete lifecycle, payload contracts, and integration trade-offs, read the Integration Guide .

Finishing a run does not blindly send every run to an LLM.

POST /runs/{run_id}/finish persists the terminal run state and duration.

A run is eligible for RAG evaluation only when all of the following are true:

• the run exists,
• its status is finished,
• the final answer is non-empty,
• at least one retrieval row exists,
• at least one retrieved chunk contains usable text or content.

If the run is not eligible, Diagent still queues rule-based anomaly detection.

For eligible runs, Diagent uses an ordered Celery chain:

anomaly detection
 -> RAG evaluation
 -> optional diagnosis decision

The RAG evaluation task uses an immutable Celery signature, so the anomaly task result is not forwarded as an accidental positional argument.

The judge produces normalized scores for:

• faithfulness
• answer_relevancy
• context_precision
• overall_score — arithmetic mean of the three metrics

Persisted judge scores must be numeric, finite, and within 0.0 <= score <= 1.0. Invalid values are rejected rather than clamped into a valid-looking score.

Diagnosis is a separate decision from both evaluation and alert generation.

With the current trigger policy, a low RAG score is required. Given a score below DIAGNOSIS_RAG_SCORE_THRESHOLD:

The diagnostician is read-only: it gathers persisted evidence and returns one root cause from the current bounded taxonomy.

Thresholds are configurable through environment variables.

Diagent can observe what happened. It cannot infer every application-specific rule about what should have happened.

Examples:

• a workflow required retrieval but retrieval never ran,
• a route allowed one operation but another operation executed,
• a known fallback condition occurred but fallback was skipped.

Those expectations belong to the application or its integration layer, where the domain evidence is available.

Keep these concepts separate:

telemetry tells Diagent what happened
application policy defines what should have happened

Also:

policy span != alert
policy span != diagnosis trigger
policy span != automatically consumed diagnosis evidence

The generic main branch does not turn arbitrary application policy spans into alerts or diagnosis evidence.

A concrete PathFinderShip reference integration is available on the example/pathfindership-integration branch. It demonstrates a deliberately bounded, application-specific extension without moving that behavior into generic main.

The FastAPI-generated schema at /docs is the authoritative interactive API reference.

Diagent is self-hosted, but main does not include built-in API authentication or tenant isolation.

Before exposing the service beyond a trusted network:

• put it behind appropriate network controls or an authenticated reverse proxy,
• restrict PostgreSQL and Redis exposure,
• use non-default credentials,
• treat telemetry as potentially sensitive application data.

Do not send:

• API keys, passwords, or authorization headers,
• raw .env content,
• image binaries or base64 payloads,
• unnecessary full documents,
• unbounded provider responses,
• fabricated cost or token values.

Prefer bounded evidence that preserves what is needed for debugging: source, rank, score, mode, status, duration, and concise error types when known.

diagent/
├── api/ # FastAPI application and routes
├── core/ # Tracer, anomaly detectors, RAG evaluation, diagnosis
├── models/ # SQLAlchemy models
├── schemas/ # Pydantic API schemas
└── workers/ # Celery app and background tasks
docs/
├── INTEGRATION_GUIDE.md
├── SCHEMA.md
└── test_worker.md
alembic/ # Database migrations
tests/ # API, detector, tracer, evaluation, diagnosis tests

A core design boundary is that the HTTP tracer has no database or ORM dependency. External Python processes can emit telemetry without importing Diagent's persistence layer.

Install runtime and test dependencies:

python -m pip install -r requirements.txt
python -m pip install -r requirements-dev.txt
python -m pip install -e .

Run the test suite:

python -m pytest -q

Check the Compose stack:

docker compose ps
docker compose logs -f api worker

Validate a local change before committing:

python -m pytest -q
git diff --check

• Integration Guide — integration methods, telemetry contracts, lifecycle, RAG eligibility, policy boundaries, and troubleshooting
• Database Schema — persisted entities and relationships
• Worker Test Notes — worker-oriented testing notes
• PathFinderShip reference branch — concrete application-specific integration example

Diagent intentionally keeps main generic. Integrations should add only the telemetry and application-specific mapping they can justify with real evidence.