Files
provenance/backend/app/integrations/models/anthropic_provider.py
T
justin 330543f9ce Fix #215: pluggable LLM + embedding provider abstraction
Adds the vendor-agnostic seam the AI assistant + match-ranking plug into:
- LLMProvider / EmbeddingProvider ABCs (base.py). LLM and embeddings are
  SEPARATE abstractions — Anthropic has no embeddings endpoint, so each is
  configured independently and either can be off.
- NullLLMProvider / NullEmbeddingProvider — the default; fail loud with a clear
  "not configured" error so AI-off deployments don't silently no-op.
- AnthropicLLMProvider — first concrete LLM impl, via the official anthropic SDK
  (default model claude-opus-4-8). A local provider (e.g. Ollama) would be
  another subclass of the same interface.
- Factory in deps.py (get_llm_provider / get_embedding_provider) selects by
  env (MODEL_PROVIDER / EMBEDDING_PROVIDER); documented in .env.example.

Providers are read-only text/vector producers — they never touch the DB, so the
"AI never writes autonomously" invariant (CLAUDE.md #1) holds; writes will go
through ChangeProposal (#214).

Tests: provider selection (null default, anthropic when keyed, fallback without
key) + null providers raise. 81 passed.

Closes #215

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: Justin Paul <justin@jpaul.me>
2026-06-09 12:51:01 -04:00

25 lines
1017 B
Python

"""Anthropic LLM provider (official SDK). Self-hosters who want everything to
stay on their own metal would configure a local provider instead (e.g. Ollama) —
that's a future implementation of the same LLMProvider interface."""
from anthropic import AsyncAnthropic
from app.integrations.models.base import LLMProvider
class AnthropicLLMProvider(LLMProvider):
def __init__(self, *, api_key: str, model: str, max_tokens: int = 4096) -> None:
self._client = AsyncAnthropic(api_key=api_key)
self._model = model
self._max_tokens = max_tokens
async def complete(self, *, prompt: str, system: str | None = None) -> str:
resp = await self._client.messages.create(
model=self._model,
max_tokens=self._max_tokens,
system=system or "",
messages=[{"role": "user", "content": prompt}],
)
# content is a list of blocks; concatenate the text ones.
return "".join(b.text for b in resp.content if b.type == "text")