feat(api): pluggable LLM provider (Ollama/Anthropic) + /ingest/json

knowledge-graph-api/.env.example

@@ -9,11 +9,19 @@ REDIS_URL=redis://redis:6379
 REDIS_INDEX_NAME=kg_vectors
 REDIS_VECTOR_DIM=768
 
+# LLM provider for extraction stage: "ollama" (default) or "anthropic".
+# Embeddings always go to Ollama regardless of this setting.
+KG_LLM_PROVIDER=ollama
+
 # Ollama (Inference API locale)
 OLLAMA_BASE_URL=http://ollama:11434
 OLLAMA_LLM_MODEL=llama3
 OLLAMA_EMBEDDING_MODEL=nomic-embed-text
 
+# Anthropic (used only when KG_LLM_PROVIDER=anthropic)
+ANTHROPIC_API_KEY=
+ANTHROPIC_EXTRACTION_MODEL=claude-haiku-4-5
+
 # Chunking
 CHUNK_SIZE=1024
 CHUNK_OVERLAP=128

knowledge-graph-api/api/main.py

@@ -72,20 +72,29 @@ async def health() -> HealthResponse:
     except Exception as exc:
         logger.warning("health_redis_fail", error=str(exc))
 
-    # Ollama
+    # LLM provider — meaning depends on KG_LLM_PROVIDER. When "anthropic"
+    # we still ping Ollama (because embeddings always run there); the
+    # field name stays "ollama" for backward compat with existing clients.
     try:
         async with httpx.AsyncClient(timeout=5.0) as client:
             resp = await client.get(f"{settings.OLLAMA_BASE_URL}/api/tags")
             ollama_ok = resp.status_code == 200
     except Exception as exc:
         logger.warning("health_ollama_fail", error=str(exc))
 
+    # When anthropic is configured as extraction provider, also surface
+    # whether the API key is present (cheap check — no network call).
+    provider = settings.KG_LLM_PROVIDER.lower().strip()
+    if provider == "anthropic" and not settings.ANTHROPIC_API_KEY:
+        logger.warning("health_anthropic_no_key")
+
     all_ok = neo4j_ok and redis_ok and ollama_ok
     return HealthResponse(
         status="healthy" if all_ok else "degraded",
         neo4j=neo4j_ok,
         redis=redis_ok,
         ollama=ollama_ok,
+        llm_provider=provider,
     )
 
 

knowledge-graph-api/api/routes/ingest.py

@@ -8,7 +8,7 @@
 
 from fastapi import APIRouter, File, Form, HTTPException, UploadFile
 
-from api.schemas import IngestRequest
+from api.schemas import IngestJsonRequest, IngestRequest
 from pipeline.ingest import IngestOptions, IngestResult, IngestionPipeline
 from utils.logger import logger
 
@@ -129,3 +129,76 @@ async def upload_and_ingest(
                 pass
 
     return result
+
+
+@router.post("/ingest/json", response_model=IngestResult)
+async def ingest_json(body: IngestJsonRequest) -> IngestResult:
+    """Ingest a structured JSON document (e.g. a news article).
+
+    Serialises the payload as a plain-text file with title, body, and
+    a trailing metadata block, then runs it through the standard
+    ingestion pipeline. The ``namespace`` field maps to ``thread_id``
+    (the pipeline's partition key).
+
+    Args:
+        body: The structured document to ingest.
+
+    Returns:
+        An ``IngestResult`` summary.
+    """
+    serialised = (
+        f"{body.title}\n\n"
+        f"{body.body}\n\n"
+        "---\n"
+        f"Source: {body.source}\n"
+        f"URL: {body.url}\n"
+        f"Published: {body.published_at.isoformat()}\n"
+    )
+
+    # NamedTemporaryFile defaults: random suffix in the OS temp dir. We
+    # deliberately do NOT derive the path from request fields — even a
+    # whitelist-sanitised body field would still trip CodeQL's
+    # "uncontrolled data in path expression" check, and the readable
+    # filename is debug-only (the URL/source live in the file body).
+    tmp_path: str | None = None
+    try:
+        with tempfile.NamedTemporaryFile(
+            delete=False,
+            suffix=".txt",
+            mode="w",
+            encoding="utf-8",
+        ) as tmp:
+            tmp.write(serialised)
+            tmp_path = tmp.name
+
+        logger.info(
+            "ingest_json_received",
+            title=body.title,
+            source=body.source,
+            namespace=body.namespace,
+            size=len(serialised),
+            tmp=tmp_path,
+        )
+
+        pipeline = IngestionPipeline()
+        try:
+            result = await pipeline.ingest(
+                file_path=tmp_path,
+                thread_id=body.namespace,
+                options=IngestOptions(skip_existing=body.skip_existing),
+            )
+        except ValueError as exc:
+            logger.error("ingest_json_error", error=str(exc))
+            raise HTTPException(status_code=400, detail=str(exc))
+        except Exception as exc:
+            logger.error("ingest_json_error", error=str(exc))
+            raise HTTPException(status_code=500, detail="Ingestion failed")
+
+    finally:
+        if tmp_path:
+            try:
+                os.unlink(tmp_path)
+            except OSError:
+                pass
+
+    return result

knowledge-graph-api/api/schemas.py

@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
-from pydantic import BaseModel
+from datetime import datetime
+
+from pydantic import BaseModel, ConfigDict, HttpUrl
 
 
 class IngestRequest(BaseModel):
@@ -13,6 +15,26 @@ class IngestRequest(BaseModel):
     skip_existing: bool = True
 
 
+class IngestJsonRequest(BaseModel):
+    """Body for POST /ingest/json — structured news ingestion.
+
+    Designed for ingesting financial news (or any short structured
+    document) without writing a file to disk on the client side. The
+    server serialises the payload to a .txt file and feeds it through
+    the standard ingestion pipeline.
+    """
+
+    model_config = ConfigDict(extra="forbid")
+
+    title: str
+    body: str
+    url: HttpUrl
+    source: str
+    published_at: datetime
+    namespace: str
+    skip_existing: bool = True
+
+
 class QueryRequest(BaseModel):
     """Body for POST /query and POST /query/stream."""
 
@@ -28,4 +50,8 @@ class HealthResponse(BaseModel):
     status: str
     neo4j: bool
     redis: bool
+    # ``ollama`` stays for backward compatibility with existing clients.
+    # When KG_LLM_PROVIDER=anthropic this field reports the Anthropic
+    # API reachability instead, and ``llm_provider`` identifies which.
     ollama: bool
+    llm_provider: str = "ollama"

knowledge-graph-api/config/settings.py

@@ -17,6 +17,12 @@ class Settings(BaseSettings):
     REDIS_INDEX_NAME: str = "kg_vectors"
     REDIS_VECTOR_DIM: int = 768
 
+    # LLM provider selection for the extraction stage.
+    # "ollama" (default) keeps the legacy local-inference behaviour.
+    # "anthropic" routes extraction to Claude (Haiku by default).
+    # Embeddings always go to Ollama regardless of this setting.
+    KG_LLM_PROVIDER: str = "ollama"
+
     # Ollama (Inference API locale)
     OLLAMA_BASE_URL: str = "http://localhost:11434"
     OLLAMA_LLM_MODEL: str = "llama3"
@@ -27,6 +33,10 @@ class Settings(BaseSettings):
     OLLAMA_EXTRACTION_MODEL: str = ""
     OLLAMA_EMBEDDING_MODEL: str = "nomic-embed-text"
 
+    # Anthropic (used when KG_LLM_PROVIDER=anthropic)
+    ANTHROPIC_API_KEY: str = ""
+    ANTHROPIC_EXTRACTION_MODEL: str = "claude-haiku-4-5"
+
     # App
     CHUNK_SIZE: int = 1024
     CHUNK_OVERLAP: int = 128

knowledge-graph-api/pipeline/extractor.py

@@ -1,16 +1,22 @@
-"""Entity and relation extraction via Ollama LLM."""
+"""Entity and relation extraction via a pluggable LLM provider.
+
+The provider (Ollama or Anthropic Claude) is resolved at construction
+time via ``pipeline.llm.get_llm_provider()`` driven by the
+``KG_LLM_PROVIDER`` setting. The extractor itself is provider-agnostic:
+it sends the system + user prompts and parses the JSON response.
+"""
 
 from __future__ import annotations
 
 import json
 
-import httpx
 from pydantic import BaseModel
 from tenacity import retry, stop_after_attempt, wait_exponential
 
 from config.settings import settings
 from models.graph_node import VALID_NODE_TYPES
 from models.relation import VALID_RELATION_TYPES
+from pipeline.llm import LLMProvider, get_llm_provider
 from utils.logger import logger
 
 # ── Result models ────────────────────────────────────────────────────
@@ -119,12 +125,10 @@ class ExtractionResult(BaseModel):
 # ── Extractor class ──────────────────────────────────────────────────
 
 class EntityExtractor:
-    """Extracts entities and relations from text chunks using Ollama."""
+    """Extracts entities and relations from text chunks via an LLM provider."""
 
-    def __init__(self) -> None:
-        self.base_url = settings.OLLAMA_BASE_URL
-        # Use dedicated extraction model if configured, otherwise fall back to LLM model.
-        self.model = settings.OLLAMA_EXTRACTION_MODEL or settings.OLLAMA_LLM_MODEL
+    def __init__(self, provider: LLMProvider | None = None) -> None:
+        self._provider = provider or get_llm_provider()
 
     async def extract(self, chunk: str, context: str = "") -> ExtractionResult:
         """Extract entities and relations from a text chunk.
@@ -163,19 +167,5 @@ async def extract(self, chunk: str, context: str = "") -> ExtractionResult:
         reraise=True,
     )
     async def _call_llm(self, user_content: str) -> str:
-        """Call the Ollama chat endpoint and return the raw message content."""
-        async with httpx.AsyncClient(timeout=120.0) as client:
-            response = await client.post(
-                f"{self.base_url}/api/chat",
-                json={
-                    "model": self.model,
-                    "messages": [
-                        {"role": "system", "content": SYSTEM_PROMPT},
-                        {"role": "user", "content": user_content},
-                    ],
-                    "stream": False,
-                    "format": "json",
-                },
-            )
-            response.raise_for_status()
-            return response.json()["message"]["content"]
+        """Delegate the chat call to the configured LLM provider."""
+        return await self._provider.chat_json(SYSTEM_PROMPT, user_content)

knowledge-graph-api/pipeline/llm/__init__.py

@@ -0,0 +1,18 @@
+"""LLM provider abstraction for entity/relation extraction.
+
+The extraction stage of the pipeline calls an LLM to convert text chunks
+into structured JSON. The provider is pluggable so the same pipeline can
+run against a local Ollama model, a hosted Anthropic Claude model, or
+future providers without touching the call sites.
+
+Embeddings are NOT covered by this abstraction — they remain wired to
+``OLLAMA_EMBEDDING_MODEL`` via ``pipeline.embedder`` because the vector
+index is sized to that model's output (768 dims for ``nomic-embed-text``).
+"""
+
+from __future__ import annotations
+
+from pipeline.llm.base import LLMProvider
+from pipeline.llm.factory import get_llm_provider
+
+__all__ = ["LLMProvider", "get_llm_provider"]

knowledge-graph-api/pipeline/llm/anthropic_provider.py

@@ -0,0 +1,72 @@
+"""Anthropic Claude provider for extraction-time JSON output.
+
+Uses the ``anthropic`` Python SDK directly (no MAF dependency) to keep
+this package lightweight. Targets Haiku by default for cost: extraction
+is a high-volume, narrow-schema task — Haiku 4.5 is the right tool.
+
+The provider forces JSON output by appending a brief reinforcement to
+the system prompt and stripping any code fences from the response.
+"""
+
+from __future__ import annotations
+
+from anthropic import AsyncAnthropic
+
+from config.settings import settings
+from pipeline.llm.base import LLMProvider
+
+_JSON_REINFORCEMENT = (
+    "\n\nIMPORTANT: respond with raw JSON only. No prose, no code fences, "
+    "no preamble — just the JSON object starting with { and ending with }."
+)
+
+
+def _strip_code_fences(text: str) -> str:
+    """Remove ```json … ``` wrappers that Claude sometimes emits anyway."""
+
+    stripped = text.strip()
+    if stripped.startswith("```"):
+        # drop the opening fence (with or without language tag)
+        first_newline = stripped.find("\n")
+        if first_newline != -1:
+            stripped = stripped[first_newline + 1 :]
+        # drop the closing fence
+        if stripped.rstrip().endswith("```"):
+            stripped = stripped.rstrip()[:-3]
+    return stripped.strip()
+
+
+class AnthropicProvider(LLMProvider):
+    """Calls Claude via the Anthropic Messages API and returns raw JSON text."""
+
+    def __init__(
+        self,
+        *,
+        api_key: str | None = None,
+        model: str | None = None,
+        max_tokens: int = 4096,
+    ) -> None:
+        resolved_key = api_key or settings.ANTHROPIC_API_KEY
+        if not resolved_key:
+            raise RuntimeError(
+                "ANTHROPIC_API_KEY is required when KG_LLM_PROVIDER=anthropic"
+            )
+        self._client = AsyncAnthropic(api_key=resolved_key)
+        self._model = model or settings.ANTHROPIC_EXTRACTION_MODEL
+        self._max_tokens = max_tokens
+
+    async def chat_json(self, system: str, user: str) -> str:
+        response = await self._client.messages.create(
+            model=self._model,
+            max_tokens=self._max_tokens,
+            system=system + _JSON_REINFORCEMENT,
+            messages=[{"role": "user", "content": user}],
+        )
+        # The Anthropic SDK returns a list of content blocks; for plain
+        # text completions there is exactly one TextBlock.
+        parts: list[str] = []
+        for block in response.content:
+            text = getattr(block, "text", None)
+            if isinstance(text, str):
+                parts.append(text)
+        return _strip_code_fences("".join(parts))

knowledge-graph-api/pipeline/llm/base.py

@@ -0,0 +1,27 @@
+"""Abstract LLM provider interface for extraction-time chat calls."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+
+
+class LLMProvider(ABC):
+    """Provider contract for the extraction LLM.
+
+    Implementations must return the raw JSON string emitted by the model
+    so the caller can ``json.loads`` it. Any parse-error tolerance lives
+    in the caller, not the provider — providers should NOT wrap parsing.
+    """
+
+    @abstractmethod
+    async def chat_json(self, system: str, user: str) -> str:
+        """Send a chat completion and return the raw response text.
+
+        Args:
+            system: System prompt (instructions + schema + examples).
+            user: User content (the text chunk to extract from).
+
+        Returns:
+            The model's raw textual output — expected to be JSON, but the
+            provider does not validate that.
+        """