Milestone 6 — Generated Answer Pipeline (AGWMP)

Overview

Milestone 6 implements the Automated Generated-answer With Memory and Pipeline (AGWMP) service — the Flow B generation pipeline that automatically produces draft answers for unanswered customer questions using LLMs and retrieval-augmented generation.

Architecture

The pipeline follows the layered architecture established in previous milestones:

API (FastAPI)
  └─ AGWMPService (application)
       ├─ WorkItemRepository (domain protocol → PostgresWorkItemRepository)
       ├─ SearchAdapter[]    (domain protocol → ConfluenceAdapter, JiraAdapter)
       ├─ VectorRepository   (domain protocol → WeaviateAdapter)
       ├─ LLMAdapter         (domain protocol → OpenAIAdapter)
       └─ RAGContextBuilder  (pure domain service)

Key Design Decisions

Known-Answer Detection (Weaviate, similarity ≥ 0.92)

Before calling the LLM, the service queries Weaviate for semantically similar approved answers. If any existing answer has a cosine similarity ≥ 0.92 to the question, it is surfaced as a GeneratedCandidate without an LLM call. This prevents duplicate generation and leverages the growing knowledge base.

No-Content Escalation (< 2 RAG sources)

If fewer than 2 RAG sources are assembled, the work item is moved to ESCALATED state and no LLM call is made. This avoids generating hallucinated answers when there is insufficient grounding material. Escalated items optionally create a Jira issue for human review.

Circuit Breaker (in-process, 5 failures / 60 s recovery)

An in-process circuit breaker wraps the LLM call. After 5 consecutive failures, the circuit opens for 60 seconds to prevent cascading timeouts. In OPEN state, new work items are immediately escalated rather than queued behind a failing LLM.

Note: The circuit breaker maintains state per-process. Production multi-worker Celery deployments should persist circuit state in Redis for shared circuit behaviour.

RAG Context Assembly

The RAGContextBuilder sorts and trims sources within a configurable token budget (default 4,000 tokens). Sources are ranked by: 1. Staleness (fresh sources first; > 12 months old = stale) 2. Authority weight (1.0 for primary docs, 0.5 for community content) 3. Relevance score (keyword overlap with the question) 4. Last-modified date (most recent first)

Token estimation uses a 4 chars ≈ 1 token heuristic to avoid a heavy ML dependency in the domain layer.

F2 Score (Recall-Weighted, β = 2)

RAG retrieval quality is evaluated using the F2 score (β = 2), which weights recall 4× more than precision. This reflects the asymmetric cost of missing a relevant source (answers may be incomplete) vs. including a marginally relevant source (over-long prompts). The quality gate is F2 ≥ 0.78.

WorkItem State Machine

NEW
 ├─→ ESCALATED (no-content / circuit open / LLM error)
 └─→ GENERATING
      ├─→ PENDING_VALIDATION (known-answer bypass or successful generation)
      └─→ ESCALATED (no-content after GENERATING)

PENDING_VALIDATION → IN_CLASSIFICATION → READY_FOR_REVIEW
                                          ├─→ VALIDATED
                                          └─→ NEEDS_REWORK

VALIDATED → CLOSED
ESCALATED → CLOSED

Files Added

Domain

File	Description
`src/raw_to_knowledge/domain/enums/generation.py`	`WorkItemState`, `CandidateOrigin`, `HornType` enums
`src/raw_to_knowledge/domain/models/generation.py`	`RAGSource`, `RAGContext`, `Citation`, `LLMResponse`, `WorkItem`, `GeneratedCandidate`, `KnownAnswerMatch`
`src/raw_to_knowledge/domain/protocols/llm.py`	`LLMAdapter` protocol
`src/raw_to_knowledge/domain/protocols/search.py`	`SearchAdapter`, `VectorRepository` protocols
`src/raw_to_knowledge/domain/services/rag_context_builder.py`	`RAGContextBuilder` — pure domain, no I/O
`src/raw_to_knowledge/domain/services/circuit_breaker.py`	`CircuitBreaker` state machine
`src/raw_to_knowledge/domain/services/f2_scorer.py`	`compute_f2_score`, `meets_quality_gate`

Infrastructure

File	Description
`src/raw_to_knowledge/infrastructure/llm/openai_adapter.py`	`OpenAIAdapter` — OpenAI-compatible LLM
`src/raw_to_knowledge/infrastructure/confluence/confluence_adapter.py`	`ConfluenceAdapter` — Confluence CQL search
`src/raw_to_knowledge/infrastructure/jira/jira_adapter.py`	`JiraAdapter` — Jira JQL search + ticket creation
`src/raw_to_knowledge/infrastructure/vector/weaviate_adapter.py`	`WeaviateAdapter` — known-answer detection + indexing
`src/raw_to_knowledge/infrastructure/celery_app.py`	Celery app instance (`raw_to_knowledge_generation` queue)
`src/raw_to_knowledge/infrastructure/persistence/models.py`	Added `WorkItemORM`, `GeneratedCandidateORM`
`src/raw_to_knowledge/infrastructure/persistence/repositories.py`	Added `PostgresWorkItemRepository`

Application

File	Description
`src/raw_to_knowledge/application/generation/agwmp_service.py`	`AGWMPService` — pipeline orchestrator
`src/raw_to_knowledge/application/generation/tasks.py`	Celery task `process_work_item`

API

File	Description
`src/raw_to_knowledge/api/v1/generation.py`	4 FastAPI routes (POST enqueue, GET list, GET single, GET candidate)

Migration

File	Description
`alembic/versions/005_generation_schema.py`	`work_items` + `generated_candidates` tables

API Endpoints

Method	Path	Description
`POST`	`/v1/generation/work-items`	Enqueue question for generation (202 Accepted)
`GET`	`/v1/generation/work-items`	List work items (optional `?state=` filter)
`GET`	`/v1/generation/work-items/{id}`	Get a single work item
`GET`	`/v1/generation/work-items/{id}/generated-candidate`	Get the generated answer

Configuration

New settings in src/raw_to_knowledge/config.py:

Setting	Default	Description
`OPENAI_API_KEY`	`""`	OpenAI (or compatible) API key
`LLM_MODEL`	`gpt-4o-mini`	Default LLM model
`RAG_TOKEN_LIMIT`	`4000`	Max token budget for RAG prompt
`WEAVIATE_URL`	`http://localhost:8080`	Weaviate URL
`WEAVIATE_API_KEY`	`""`	Weaviate API key (empty = unauthenticated)
`CONFLUENCE_BASE_URL`	`""`	Confluence base URL
`CONFLUENCE_USERNAME`	`""`	Atlassian email
`CONFLUENCE_API_TOKEN`	`""`	Atlassian API token
`CONFLUENCE_SPACE_KEYS`	`""`	Comma-separated space keys
`JIRA_BASE_URL`	`""`	Jira base URL
`JIRA_USERNAME`	`""`	Atlassian email
`JIRA_API_TOKEN`	`""`	Atlassian API token
`JIRA_PROJECT_KEY`	`""`	Default project for escalation tickets

Quality Gates

Gate	Threshold	Measurement
RAG F2 score	≥ 0.78	Recall-weighted F2 over golden question set
Known-answer detection	Similarity ≥ 0.92	Weaviate cosine similarity threshold
No-content escalation	< 2 sources	RAG source count after assembly

Running the Worker

# Start the Celery worker
celery -A raw_to_knowledge.infrastructure.celery_app worker --loglevel=info -Q raw_to_knowledge_generation

# Or with concurrency
celery -A raw_to_knowledge.infrastructure.celery_app worker --loglevel=info -Q raw_to_knowledge_generation --concurrency=4

Testing

# Unit tests (no external deps)
pytest tests/unit/test_rag_context_builder.py tests/unit/test_f2_scorer.py \
       tests/unit/test_circuit_breaker.py tests/unit/test_agwmp_service.py

# Integration tests (requires DATABASE_URL)
DATABASE_URL=postgresql+asyncpg://... pytest tests/integration/test_generation_roundtrip.py -m integration

# RAG quality evaluation (requires rag_labels.jsonl)
python tests/golden/evaluate.py --rag-labels tests/golden/rag_labels.jsonl