Skip to content

Commit 52f6b7e

Browse files
authored
feat(api): pluggable LLM provider (Ollama/Anthropic) + /ingest/json (#2)
* feat(api): pluggable LLM provider (Ollama/Anthropic) + /ingest/json Extraction-time LLM is now decoupled from the legacy Ollama-only path through a small ``LLMProvider`` ABC under ``pipeline/llm/``. The active provider is selected by ``KG_LLM_PROVIDER`` (default ``ollama``); set it to ``anthropic`` to route extraction to Claude Haiku via the official ``anthropic`` SDK. Embeddings stay on Ollama unconditionally — the vector index is sized for ``nomic-embed-text``. Also adds ``POST /ingest/json`` for ingesting structured news articles without writing files on the client side. The endpoint serialises ``{title, body, url, source, published_at, namespace}`` to a temp .txt and feeds it through the existing pipeline (no router changes needed — .txt is already supported). Tests: 14 new tests (extractor refactor, both providers, factory, JSON ingest serialisation contract). Full suite 23/23 green. * fix(api): drop user-derived prefix from /ingest/json tempfile CodeQL flagged the (whitelist-sanitised) safe_stem as "uncontrolled data in path expression". The readable-filename feature was debug ergonomics, not a contract — the URL and source already live inside the file body. Use NamedTemporaryFile's random default name instead. Tests untouched (the contract is the file contents, not the path).
1 parent 4bfb103 commit 52f6b7e

15 files changed

Lines changed: 586 additions & 48 deletions

knowledge-graph-api/.env.example

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,11 +9,19 @@ REDIS_URL=redis://redis:6379
99
REDIS_INDEX_NAME=kg_vectors
1010
REDIS_VECTOR_DIM=768
1111

12+
# LLM provider for extraction stage: "ollama" (default) or "anthropic".
13+
# Embeddings always go to Ollama regardless of this setting.
14+
KG_LLM_PROVIDER=ollama
15+
1216
# Ollama (Inference API locale)
1317
OLLAMA_BASE_URL=http://ollama:11434
1418
OLLAMA_LLM_MODEL=llama3
1519
OLLAMA_EMBEDDING_MODEL=nomic-embed-text
1620

21+
# Anthropic (used only when KG_LLM_PROVIDER=anthropic)
22+
ANTHROPIC_API_KEY=
23+
ANTHROPIC_EXTRACTION_MODEL=claude-haiku-4-5
24+
1725
# Chunking
1826
CHUNK_SIZE=1024
1927
CHUNK_OVERLAP=128

knowledge-graph-api/api/main.py

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -72,20 +72,29 @@ async def health() -> HealthResponse:
7272
except Exception as exc:
7373
logger.warning("health_redis_fail", error=str(exc))
7474

75-
# Ollama
75+
# LLM provider — meaning depends on KG_LLM_PROVIDER. When "anthropic"
76+
# we still ping Ollama (because embeddings always run there); the
77+
# field name stays "ollama" for backward compat with existing clients.
7678
try:
7779
async with httpx.AsyncClient(timeout=5.0) as client:
7880
resp = await client.get(f"{settings.OLLAMA_BASE_URL}/api/tags")
7981
ollama_ok = resp.status_code == 200
8082
except Exception as exc:
8183
logger.warning("health_ollama_fail", error=str(exc))
8284

85+
# When anthropic is configured as extraction provider, also surface
86+
# whether the API key is present (cheap check — no network call).
87+
provider = settings.KG_LLM_PROVIDER.lower().strip()
88+
if provider == "anthropic" and not settings.ANTHROPIC_API_KEY:
89+
logger.warning("health_anthropic_no_key")
90+
8391
all_ok = neo4j_ok and redis_ok and ollama_ok
8492
return HealthResponse(
8593
status="healthy" if all_ok else "degraded",
8694
neo4j=neo4j_ok,
8795
redis=redis_ok,
8896
ollama=ollama_ok,
97+
llm_provider=provider,
8998
)
9099

91100

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

Lines changed: 74 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88

99
from fastapi import APIRouter, File, Form, HTTPException, UploadFile
1010

11-
from api.schemas import IngestRequest
11+
from api.schemas import IngestJsonRequest, IngestRequest
1212
from pipeline.ingest import IngestOptions, IngestResult, IngestionPipeline
1313
from utils.logger import logger
1414

@@ -129,3 +129,76 @@ async def upload_and_ingest(
129129
pass
130130

131131
return result
132+
133+
134+
@router.post("/ingest/json", response_model=IngestResult)
135+
async def ingest_json(body: IngestJsonRequest) -> IngestResult:
136+
"""Ingest a structured JSON document (e.g. a news article).
137+
138+
Serialises the payload as a plain-text file with title, body, and
139+
a trailing metadata block, then runs it through the standard
140+
ingestion pipeline. The ``namespace`` field maps to ``thread_id``
141+
(the pipeline's partition key).
142+
143+
Args:
144+
body: The structured document to ingest.
145+
146+
Returns:
147+
An ``IngestResult`` summary.
148+
"""
149+
serialised = (
150+
f"{body.title}\n\n"
151+
f"{body.body}\n\n"
152+
"---\n"
153+
f"Source: {body.source}\n"
154+
f"URL: {body.url}\n"
155+
f"Published: {body.published_at.isoformat()}\n"
156+
)
157+
158+
# NamedTemporaryFile defaults: random suffix in the OS temp dir. We
159+
# deliberately do NOT derive the path from request fields — even a
160+
# whitelist-sanitised body field would still trip CodeQL's
161+
# "uncontrolled data in path expression" check, and the readable
162+
# filename is debug-only (the URL/source live in the file body).
163+
tmp_path: str | None = None
164+
try:
165+
with tempfile.NamedTemporaryFile(
166+
delete=False,
167+
suffix=".txt",
168+
mode="w",
169+
encoding="utf-8",
170+
) as tmp:
171+
tmp.write(serialised)
172+
tmp_path = tmp.name
173+
174+
logger.info(
175+
"ingest_json_received",
176+
title=body.title,
177+
source=body.source,
178+
namespace=body.namespace,
179+
size=len(serialised),
180+
tmp=tmp_path,
181+
)
182+
183+
pipeline = IngestionPipeline()
184+
try:
185+
result = await pipeline.ingest(
186+
file_path=tmp_path,
187+
thread_id=body.namespace,
188+
options=IngestOptions(skip_existing=body.skip_existing),
189+
)
190+
except ValueError as exc:
191+
logger.error("ingest_json_error", error=str(exc))
192+
raise HTTPException(status_code=400, detail=str(exc))
193+
except Exception as exc:
194+
logger.error("ingest_json_error", error=str(exc))
195+
raise HTTPException(status_code=500, detail="Ingestion failed")
196+
197+
finally:
198+
if tmp_path:
199+
try:
200+
os.unlink(tmp_path)
201+
except OSError:
202+
pass
203+
204+
return result

knowledge-graph-api/api/schemas.py

Lines changed: 27 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,9 @@
22

33
from __future__ import annotations
44

5-
from pydantic import BaseModel
5+
from datetime import datetime
6+
7+
from pydantic import BaseModel, ConfigDict, HttpUrl
68

79

810
class IngestRequest(BaseModel):
@@ -13,6 +15,26 @@ class IngestRequest(BaseModel):
1315
skip_existing: bool = True
1416

1517

18+
class IngestJsonRequest(BaseModel):
19+
"""Body for POST /ingest/json — structured news ingestion.
20+
21+
Designed for ingesting financial news (or any short structured
22+
document) without writing a file to disk on the client side. The
23+
server serialises the payload to a .txt file and feeds it through
24+
the standard ingestion pipeline.
25+
"""
26+
27+
model_config = ConfigDict(extra="forbid")
28+
29+
title: str
30+
body: str
31+
url: HttpUrl
32+
source: str
33+
published_at: datetime
34+
namespace: str
35+
skip_existing: bool = True
36+
37+
1638
class QueryRequest(BaseModel):
1739
"""Body for POST /query and POST /query/stream."""
1840

@@ -28,4 +50,8 @@ class HealthResponse(BaseModel):
2850
status: str
2951
neo4j: bool
3052
redis: bool
53+
# ``ollama`` stays for backward compatibility with existing clients.
54+
# When KG_LLM_PROVIDER=anthropic this field reports the Anthropic
55+
# API reachability instead, and ``llm_provider`` identifies which.
3156
ollama: bool
57+
llm_provider: str = "ollama"

knowledge-graph-api/config/settings.py

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,12 @@ class Settings(BaseSettings):
1717
REDIS_INDEX_NAME: str = "kg_vectors"
1818
REDIS_VECTOR_DIM: int = 768
1919

20+
# LLM provider selection for the extraction stage.
21+
# "ollama" (default) keeps the legacy local-inference behaviour.
22+
# "anthropic" routes extraction to Claude (Haiku by default).
23+
# Embeddings always go to Ollama regardless of this setting.
24+
KG_LLM_PROVIDER: str = "ollama"
25+
2026
# Ollama (Inference API locale)
2127
OLLAMA_BASE_URL: str = "http://localhost:11434"
2228
OLLAMA_LLM_MODEL: str = "llama3"
@@ -27,6 +33,10 @@ class Settings(BaseSettings):
2733
OLLAMA_EXTRACTION_MODEL: str = ""
2834
OLLAMA_EMBEDDING_MODEL: str = "nomic-embed-text"
2935

36+
# Anthropic (used when KG_LLM_PROVIDER=anthropic)
37+
ANTHROPIC_API_KEY: str = ""
38+
ANTHROPIC_EXTRACTION_MODEL: str = "claude-haiku-4-5"
39+
3040
# App
3141
CHUNK_SIZE: int = 1024
3242
CHUNK_OVERLAP: int = 128

knowledge-graph-api/pipeline/extractor.py

Lines changed: 13 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -1,16 +1,22 @@
1-
"""Entity and relation extraction via Ollama LLM."""
1+
"""Entity and relation extraction via a pluggable LLM provider.
2+
3+
The provider (Ollama or Anthropic Claude) is resolved at construction
4+
time via ``pipeline.llm.get_llm_provider()`` driven by the
5+
``KG_LLM_PROVIDER`` setting. The extractor itself is provider-agnostic:
6+
it sends the system + user prompts and parses the JSON response.
7+
"""
28

39
from __future__ import annotations
410

511
import json
612

7-
import httpx
813
from pydantic import BaseModel
914
from tenacity import retry, stop_after_attempt, wait_exponential
1015

1116
from config.settings import settings
1217
from models.graph_node import VALID_NODE_TYPES
1318
from models.relation import VALID_RELATION_TYPES
19+
from pipeline.llm import LLMProvider, get_llm_provider
1420
from utils.logger import logger
1521

1622
# ── Result models ────────────────────────────────────────────────────
@@ -119,12 +125,10 @@ class ExtractionResult(BaseModel):
119125
# ── Extractor class ──────────────────────────────────────────────────
120126

121127
class EntityExtractor:
122-
"""Extracts entities and relations from text chunks using Ollama."""
128+
"""Extracts entities and relations from text chunks via an LLM provider."""
123129

124-
def __init__(self) -> None:
125-
self.base_url = settings.OLLAMA_BASE_URL
126-
# Use dedicated extraction model if configured, otherwise fall back to LLM model.
127-
self.model = settings.OLLAMA_EXTRACTION_MODEL or settings.OLLAMA_LLM_MODEL
130+
def __init__(self, provider: LLMProvider | None = None) -> None:
131+
self._provider = provider or get_llm_provider()
128132

129133
async def extract(self, chunk: str, context: str = "") -> ExtractionResult:
130134
"""Extract entities and relations from a text chunk.
@@ -163,19 +167,5 @@ async def extract(self, chunk: str, context: str = "") -> ExtractionResult:
163167
reraise=True,
164168
)
165169
async def _call_llm(self, user_content: str) -> str:
166-
"""Call the Ollama chat endpoint and return the raw message content."""
167-
async with httpx.AsyncClient(timeout=120.0) as client:
168-
response = await client.post(
169-
f"{self.base_url}/api/chat",
170-
json={
171-
"model": self.model,
172-
"messages": [
173-
{"role": "system", "content": SYSTEM_PROMPT},
174-
{"role": "user", "content": user_content},
175-
],
176-
"stream": False,
177-
"format": "json",
178-
},
179-
)
180-
response.raise_for_status()
181-
return response.json()["message"]["content"]
170+
"""Delegate the chat call to the configured LLM provider."""
171+
return await self._provider.chat_json(SYSTEM_PROMPT, user_content)
Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
"""LLM provider abstraction for entity/relation extraction.
2+
3+
The extraction stage of the pipeline calls an LLM to convert text chunks
4+
into structured JSON. The provider is pluggable so the same pipeline can
5+
run against a local Ollama model, a hosted Anthropic Claude model, or
6+
future providers without touching the call sites.
7+
8+
Embeddings are NOT covered by this abstraction — they remain wired to
9+
``OLLAMA_EMBEDDING_MODEL`` via ``pipeline.embedder`` because the vector
10+
index is sized to that model's output (768 dims for ``nomic-embed-text``).
11+
"""
12+
13+
from __future__ import annotations
14+
15+
from pipeline.llm.base import LLMProvider
16+
from pipeline.llm.factory import get_llm_provider
17+
18+
__all__ = ["LLMProvider", "get_llm_provider"]
Lines changed: 72 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,72 @@
1+
"""Anthropic Claude provider for extraction-time JSON output.
2+
3+
Uses the ``anthropic`` Python SDK directly (no MAF dependency) to keep
4+
this package lightweight. Targets Haiku by default for cost: extraction
5+
is a high-volume, narrow-schema task — Haiku 4.5 is the right tool.
6+
7+
The provider forces JSON output by appending a brief reinforcement to
8+
the system prompt and stripping any code fences from the response.
9+
"""
10+
11+
from __future__ import annotations
12+
13+
from anthropic import AsyncAnthropic
14+
15+
from config.settings import settings
16+
from pipeline.llm.base import LLMProvider
17+
18+
_JSON_REINFORCEMENT = (
19+
"\n\nIMPORTANT: respond with raw JSON only. No prose, no code fences, "
20+
"no preamble — just the JSON object starting with { and ending with }."
21+
)
22+
23+
24+
def _strip_code_fences(text: str) -> str:
25+
"""Remove ```json … ``` wrappers that Claude sometimes emits anyway."""
26+
27+
stripped = text.strip()
28+
if stripped.startswith("```"):
29+
# drop the opening fence (with or without language tag)
30+
first_newline = stripped.find("\n")
31+
if first_newline != -1:
32+
stripped = stripped[first_newline + 1 :]
33+
# drop the closing fence
34+
if stripped.rstrip().endswith("```"):
35+
stripped = stripped.rstrip()[:-3]
36+
return stripped.strip()
37+
38+
39+
class AnthropicProvider(LLMProvider):
40+
"""Calls Claude via the Anthropic Messages API and returns raw JSON text."""
41+
42+
def __init__(
43+
self,
44+
*,
45+
api_key: str | None = None,
46+
model: str | None = None,
47+
max_tokens: int = 4096,
48+
) -> None:
49+
resolved_key = api_key or settings.ANTHROPIC_API_KEY
50+
if not resolved_key:
51+
raise RuntimeError(
52+
"ANTHROPIC_API_KEY is required when KG_LLM_PROVIDER=anthropic"
53+
)
54+
self._client = AsyncAnthropic(api_key=resolved_key)
55+
self._model = model or settings.ANTHROPIC_EXTRACTION_MODEL
56+
self._max_tokens = max_tokens
57+
58+
async def chat_json(self, system: str, user: str) -> str:
59+
response = await self._client.messages.create(
60+
model=self._model,
61+
max_tokens=self._max_tokens,
62+
system=system + _JSON_REINFORCEMENT,
63+
messages=[{"role": "user", "content": user}],
64+
)
65+
# The Anthropic SDK returns a list of content blocks; for plain
66+
# text completions there is exactly one TextBlock.
67+
parts: list[str] = []
68+
for block in response.content:
69+
text = getattr(block, "text", None)
70+
if isinstance(text, str):
71+
parts.append(text)
72+
return _strip_code_fences("".join(parts))
Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,27 @@
1+
"""Abstract LLM provider interface for extraction-time chat calls."""
2+
3+
from __future__ import annotations
4+
5+
from abc import ABC, abstractmethod
6+
7+
8+
class LLMProvider(ABC):
9+
"""Provider contract for the extraction LLM.
10+
11+
Implementations must return the raw JSON string emitted by the model
12+
so the caller can ``json.loads`` it. Any parse-error tolerance lives
13+
in the caller, not the provider — providers should NOT wrap parsing.
14+
"""
15+
16+
@abstractmethod
17+
async def chat_json(self, system: str, user: str) -> str:
18+
"""Send a chat completion and return the raw response text.
19+
20+
Args:
21+
system: System prompt (instructions + schema + examples).
22+
user: User content (the text chunk to extract from).
23+
24+
Returns:
25+
The model's raw textual output — expected to be JSON, but the
26+
provider does not validate that.
27+
"""

0 commit comments

Comments
 (0)