diff --git a/agentforge/agents/checkpoint.py b/agentforge/agents/checkpoint.py index 9226a39..410fe55 100644 --- a/agentforge/agents/checkpoint.py +++ b/agentforge/agents/checkpoint.py @@ -9,16 +9,10 @@ from functools import lru_cache -from agentforge.config import get_settings +from agentforge.config import get_settings, libpq_url - -def _conninfo(database_url: str) -> str: - """Turn a SQLAlchemy URL into a libpq conninfo string. - - The app uses ``postgresql+psycopg://…`` for langchain/SQLAlchemy; psycopg's - pool wants a plain ``postgresql://…``. - """ - return database_url.replace("postgresql+psycopg://", "postgresql://", 1) +# Kept for back-compat with existing imports/tests; delegates to the shared helper. +_conninfo = libpq_url @lru_cache diff --git a/agentforge/api/main.py b/agentforge/api/main.py index c432b66..0ede8c4 100644 --- a/agentforge/api/main.py +++ b/agentforge/api/main.py @@ -3,6 +3,7 @@ Endpoints: - ``GET /health`` liveness probe. - ``GET /metrics`` Prometheus metrics (HTTP + domain counters). +- ``GET /documents`` list ingested source documents + chunk counts. - ``GET /approvals`` list runs paused awaiting human approval. - ``POST /chat`` run the agent; may pause for approval. - ``POST /chat/stream`` stream the answer token-by-token over SSE. @@ -32,6 +33,7 @@ ApprovalRequest, ChatRequest, ChatResponse, + DocumentSummaryItem, PendingAction, PendingApprovalItem, ) @@ -138,6 +140,16 @@ def _record_domain_metrics(resp: ChatResponse) -> None: metrics.answers_total.labels(grounded).inc() +@app.get("/documents", response_model=list[DocumentSummaryItem]) +def documents() -> list[DocumentSummaryItem]: + from agentforge.rag.catalog import list_documents + + return [ + DocumentSummaryItem(source=d.source, title=d.title, chunks=d.chunks) + for d in list_documents() + ] + + @app.get("/approvals", response_model=list[PendingApprovalItem]) def approval_queue() -> list[PendingApprovalItem]: return [ diff --git a/agentforge/api/schemas.py b/agentforge/api/schemas.py index 29a0e2f..882885c 100644 --- a/agentforge/api/schemas.py +++ b/agentforge/api/schemas.py @@ -39,3 +39,9 @@ class PendingApprovalItem(BaseModel): question: str created_at: float action: PendingAction + + +class DocumentSummaryItem(BaseModel): + source: str + title: str + chunks: int diff --git a/agentforge/config.py b/agentforge/config.py index eb4a871..fdebc82 100644 --- a/agentforge/config.py +++ b/agentforge/config.py @@ -91,3 +91,12 @@ class Settings(BaseSettings): def get_settings() -> Settings: """Cached singleton — import this everywhere instead of constructing Settings().""" return Settings() + + +def libpq_url(database_url: str) -> str: + """SQLAlchemy URL -> libpq conninfo. + + The app uses ``postgresql+psycopg://…`` for langchain/SQLAlchemy; psycopg and + its pool want a plain ``postgresql://…``. Only the scheme is rewritten. + """ + return database_url.replace("postgresql+psycopg://", "postgresql://", 1) diff --git a/agentforge/rag/catalog.py b/agentforge/rag/catalog.py new file mode 100644 index 0000000..4c54636 --- /dev/null +++ b/agentforge/rag/catalog.py @@ -0,0 +1,51 @@ +"""Read-only catalog of what's ingested in the vector store. + +Powers the console's Knowledge view so RAG grounding is transparent. Queries the +``langchain_postgres`` tables directly (grouping chunks by their ``source`` +metadata) rather than embedding a search, so the listing is exact. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +from agentforge.config import get_settings, libpq_url + +# Group the collection's chunks by source document. Joins the embedding rows to +# their collection by name so we only count this app's collection. +_SQL = """ +SELECT + e.cmetadata ->> 'source' AS source, + max(e.cmetadata ->> 'title') AS title, + count(*) AS chunks +FROM langchain_pg_embedding e +JOIN langchain_pg_collection c ON c.uuid = e.collection_id +WHERE c.name = %s +GROUP BY e.cmetadata ->> 'source' +ORDER BY source +""" + + +@dataclass +class DocumentSummary: + source: str + title: str + chunks: int + + +def list_documents() -> list[DocumentSummary]: + """One row per ingested source document, or ``[]`` if the store is unreachable.""" + import psycopg + + settings = get_settings() + try: + with psycopg.connect(libpq_url(settings.database_url), connect_timeout=3) as conn: + rows = conn.execute(_SQL, (settings.collection_name,)).fetchall() + except Exception: + # Store not provisioned yet / unreachable — empty catalog, not an error. + return [] + + return [ + DocumentSummary(source=row[0] or "(unknown)", title=row[1] or "", chunks=row[2]) + for row in rows + ] diff --git a/apps/console/src/app/agent.service.ts b/apps/console/src/app/agent.service.ts index c27e52f..e400ae5 100644 --- a/apps/console/src/app/agent.service.ts +++ b/apps/console/src/app/agent.service.ts @@ -34,6 +34,12 @@ export interface PendingApprovalItem { action: PendingAction; } +export interface DocumentSummaryItem { + source: string; + title: string; + chunks: number; +} + export type StreamEvent = | { type: 'thread'; threadId: string } | { type: 'token'; text: string } @@ -68,6 +74,11 @@ export class AgentService { return this.http.get(`${API_BASE}/approvals`); } + /** Source documents ingested into the vector store. */ + documents(): Observable { + return this.http.get(`${API_BASE}/documents`); + } + /** * Stream a chat response over SSE. Uses fetch (not EventSource) because the * endpoint is a POST. Emits typed events; unsubscribing aborts the request. diff --git a/apps/console/src/app/app.component.ts b/apps/console/src/app/app.component.ts index a472748..d1c8ee1 100644 --- a/apps/console/src/app/app.component.ts +++ b/apps/console/src/app/app.component.ts @@ -14,6 +14,7 @@ import { RouterLink, RouterLinkActive, RouterOutlet } from '@angular/router'; Chat Approvals + Knowledge Operations diff --git a/apps/console/src/app/app.routes.ts b/apps/console/src/app/app.routes.ts index 7a49836..6878fc0 100644 --- a/apps/console/src/app/app.routes.ts +++ b/apps/console/src/app/app.routes.ts @@ -1,11 +1,13 @@ import { Routes } from '@angular/router'; import { ApprovalsComponent } from './approvals.component'; import { ChatComponent } from './chat.component'; +import { KnowledgeComponent } from './knowledge.component'; import { MetricsComponent } from './metrics.component'; export const routes: Routes = [ { path: '', component: ChatComponent, title: 'AgentForge — Chat' }, { path: 'approvals', component: ApprovalsComponent, title: 'AgentForge — Approvals' }, + { path: 'knowledge', component: KnowledgeComponent, title: 'AgentForge — Knowledge' }, { path: 'ops', component: MetricsComponent, title: 'AgentForge — Operations' }, { path: '**', redirectTo: '' }, ]; diff --git a/apps/console/src/app/knowledge.component.ts b/apps/console/src/app/knowledge.component.ts new file mode 100644 index 0000000..16557f3 --- /dev/null +++ b/apps/console/src/app/knowledge.component.ts @@ -0,0 +1,108 @@ +import { Component, OnInit } from '@angular/core'; +import { CommonModule } from '@angular/common'; +import { AgentService, DocumentSummaryItem } from './agent.service'; + +@Component({ + selector: 'app-knowledge', + standalone: true, + imports: [CommonModule], + template: ` +
+

Knowledge base

+ + {{ docs.length }} sources · {{ totalChunks }} chunks + +
+ +
⚠️ {{ error }}
+ +
+ Nothing ingested yet. Run the ingest job (or boot with auto-ingest) to load + the corpus the agent retrieves from. +
+ + + + + + + + + + + + + + + + +
SourceTitleChunks
{{ d.source }}{{ d.title || '—' }}{{ d.chunks }}
+ `, + styles: [ + ` + .ops-head { + display: flex; + align-items: baseline; + justify-content: space-between; + } + .muted { + font-size: 0.8rem; + color: #718096; + } + .card.empty { + color: #718096; + } + table.docs { + width: 100%; + border-collapse: collapse; + background: #fff; + border: 1px solid #e2e8f0; + border-radius: 12px; + overflow: hidden; + margin-top: 12px; + } + .docs th, + .docs td { + text-align: left; + padding: 10px 12px; + border-bottom: 1px solid #edf2f7; + font-size: 0.9rem; + } + .docs th { + background: #f7fafc; + color: #4a5568; + font-weight: 600; + } + .docs .num { + text-align: right; + } + .docs tr:last-child td { + border-bottom: 0; + } + `, + ], +}) +export class KnowledgeComponent implements OnInit { + docs: DocumentSummaryItem[] = []; + loaded = false; + error = ''; + + constructor(private agent: AgentService) {} + + get totalChunks(): number { + return this.docs.reduce((acc, d) => acc + d.chunks, 0); + } + + ngOnInit(): void { + this.agent.documents().subscribe({ + next: (docs) => { + this.docs = docs; + this.loaded = true; + }, + error: (e) => { + this.error = `Couldn't load the knowledge base: ${e?.message ?? e}`; + this.loaded = true; + }, + }); + } +} diff --git a/docs/SMOKE_TEST.md b/docs/SMOKE_TEST.md index b8e87c3..3a490b2 100644 --- a/docs/SMOKE_TEST.md +++ b/docs/SMOKE_TEST.md @@ -142,6 +142,8 @@ Open . tool + args); approving/rejecting it there clears it from the queue. - [ ] Stop the API (`docker compose stop api`) and send a message — the UI shows an error rather than hanging. +- [ ] Click **Knowledge** — a table lists the ingested sources (AML/KYC policy, + product sheet) with their chunk counts. - [ ] Click **Operations** — tiles show live counts (chat requests, grounded %, approvals, PII redactions, HTTP requests) that update as you use Chat. diff --git a/tests/test_catalog.py b/tests/test_catalog.py new file mode 100644 index 0000000..491d8c5 --- /dev/null +++ b/tests/test_catalog.py @@ -0,0 +1,21 @@ +"""Knowledge-base catalog + /documents endpoint. No DB available in unit CI, so +this exercises the graceful-empty path (store unreachable -> []).""" + +from __future__ import annotations + +from fastapi.testclient import TestClient + +from agentforge.api.main import app +from agentforge.rag.catalog import list_documents + + +def test_list_documents_empty_when_store_unreachable(): + # No Postgres in the unit-test job — must degrade to an empty catalog, + # never raise. + assert list_documents() == [] + + +def test_documents_endpoint_returns_list(): + resp = TestClient(app).get("/documents") + assert resp.status_code == 200 + assert resp.json() == []