curo/kite
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(07-03): AnthropicProvider singleton + output_config + truncation — D-03/D-07/D-12/D-13

Browse Source 
- Delete MAX_AI_CHARS constant and _client() method from anthropic_provider.py
- Add self._client = AsyncAnthropic(...) singleton in __init__ (D-07)
- Add _truncate() with 60/40 split using self._context_chars (D-13)
- Add _CLASSIFICATION_SCHEMA and _SUGGESTIONS_SCHEMA module-level constants
- classify() and suggest_topics() pass output_config with json_schema format (D-03)
- stop_reason != "end_turn" degrades to parse_classification("") (T-07-08)
- Widen __init__ signature to (api_key, model, context_chars, base_url) (uniform ctor)
- Update ai/__init__.py to pass context_chars + base_url to AnthropicProvider
- Promote test_anthropic_structured_output + add test_anthropic_stop_reason_fallback
This commit is contained in:
curo1305
2026-06-04 19:10:05 +02:00
parent ea8df02491
commit efc177a155
3 changed files with 178 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/ai/__init__.py
+5 -3
View File
@@ -19,7 +19,7 @@ from ai.provider_config import ProviderConfig, PROVIDER_DEFAULTS, SUPPORTS_JSON_
# Registry: maps provider_id → provider class
# "openai" uses OpenAIProvider (no response_format override needed — plain OpenAI)
# "anthropic" uses AnthropicProvider (native output_config, no base_url ctor arg until Plan 03)
# "anthropic" uses AnthropicProvider (native output_config, Plan 03 widened ctor accepts context_chars+base_url)
# All 8 OpenAI-compat vendors use GenericOpenAIProvider (D-16/D-17/D-18)
_REGISTRY: dict[str, type[AIProvider]] = {
"openai": OpenAIProvider,
@@ -65,11 +65,13 @@ def get_provider(config: ProviderConfig) -> AIProvider:
effective_context_chars = config.context_chars or defaults["context_chars"]
if config.provider_id == "anthropic":
# AnthropicProvider does not accept base_url until Plan 03 refactors it;
# pass only the args its current __init__ accepts.
# AnthropicProvider accepts context_chars and base_url (Plan 03 widened ctor).
# base_url is accepted for uniform factory signature but unused by the SDK.
return cls(
api_key=effective_api_key,
model=effective_model,
context_chars=effective_context_chars,
base_url=effective_base_url,
)
elif cls is GenericOpenAIProvider:
return cls(
backend/ai/anthropic_provider.py
+96 -17
View File
@@ -1,17 +1,83 @@
"""Anthropic AI provider — singleton client, output_config structured output, smart truncation.
D-03: Uses output_config={"format": {"type": "json_schema", "schema": ...}} with constrained
decoding available in anthropic SDK >=0.95.0 (GA, no beta headers needed).
D-07: self._client = AsyncAnthropic(...) created once in __init__ and reused — never recreated
per API call to preserve the httpx connection pool.
D-12/D-13: Global char constant removed; uses self._context_chars with 60/40 smart truncation.
Security: api_key is accepted from the caller (loaded from system_settings by ai_config.py
and decrypted before being passed here). The key is never stored beyond this instance's
lifetime. T-07-06 mitigated: this class never reads the api_key from env vars directly.
"""
from __future__ import annotations
import anthropic
from ai.base import AIProvider, ClassificationResult
from ai.utils import parse_classification, parse_suggestions
MAX_AI_CHARS = 8_000
# ── Output schemas for constrained decoding (D-03 / RESEARCH.md) ────────────────────────────
# additionalProperties=False required by Anthropic output_config grammar.
# "reasoning" is intentionally absent from "required" so legacy prompts that don't include it
# still produce valid JSON (Anthropic will emit it because it is declared in properties, but
# we do not enforce it in the schema to avoid refusal on minimal responses).
_CLASSIFICATION_SCHEMA: dict = {
"type": "object",
"properties": {
"assigned_topics": {"type": "array", "items": {"type": "string"}},
"new_topic_suggestions": {"type": "array", "items": {"type": "string"}},
"reasoning": {"type": "string"},
},
"required": ["assigned_topics", "new_topic_suggestions"],
"additionalProperties": False,
}
_SUGGESTIONS_SCHEMA: dict = {
"type": "object",
"properties": {
"suggested_topics": {"type": "array", "items": {"type": "string"}},
},
"required": ["suggested_topics"],
"additionalProperties": False,
}
class AnthropicProvider(AIProvider):
def __init__(self, api_key: str, model: str = "claude-sonnet-4-6"):
"""Anthropic Claude provider with singleton client and output_config structured output.
Constructor signature matches the uniform factory contract in ai/__init__.py:
api_key, model, context_chars, base_url (accepted but unused — Anthropic SDK
manages the endpoint; widened so get_provider() can call all providers uniformly).
"""
def __init__(
self,
api_key: str,
model: str = "claude-sonnet-4-6",
context_chars: int = 180_000,
base_url: str | None = None, # accepted for uniform factory signature; unused
):
self._api_key = api_key
self._model = model
self._context_chars = context_chars
# Singleton: created once in __init__, reused for all calls on this instance.
# Do NOT recreate per API call — AsyncAnthropic wraps an httpx.AsyncClient
# that maintains a connection pool; recreating per call destroys pool reuse
# and forces a new TLS handshake per request (D-07 / RESEARCH.md).
self._client = anthropic.AsyncAnthropic(api_key=self._api_key)
def _client(self):
return anthropic.AsyncAnthropic(api_key=self._api_key)
def _truncate(self, text: str) -> str:
"""D-13 smart truncation: first 60% + last 40% of context window.
Captures both document introduction and conclusion, which carry the
most topic signal for long documents.
"""
if len(text) <= self._context_chars:
return text
head_len = int(self._context_chars * 0.6)
tail_len = self._context_chars - head_len
return text[:head_len] + "\n[...truncated...]\n" + text[-tail_len:]
async def classify(
self,
@@ -22,16 +88,23 @@ class AnthropicProvider(AIProvider):
topics_str = ", ".join(existing_topics) if existing_topics else "(none yet)"
user_msg = (
f"Existing topics: [{topics_str}]\n\n"
f"Document text:\n{document_text[:MAX_AI_CHARS]}"
f"Document text:\n{self._truncate(document_text)}"
)
client = self._client()
response = await client.messages.create(
response = await self._client.messages.create(
model=self._model,
max_tokens=1024,
system=system_prompt,
messages=[{"role": "user", "content": user_msg}],
output_config={"format": {"type": "json_schema", "schema": _CLASSIFICATION_SCHEMA}},
)
raw = response.content[0].text
# Graceful degradation (T-07-08): when stop_reason is "refusal" or "max_tokens"
# the constrained decoding did not complete — fall back to parse_classification("")
# which returns an empty ClassificationResult rather than raising an exception.
stop_reason = getattr(response, "stop_reason", "end_turn")
if response.content and stop_reason == "end_turn":
raw = response.content[0].text
else:
raw = ""
return parse_classification(raw)
async def suggest_topics(
@@ -42,28 +115,34 @@ class AnthropicProvider(AIProvider):
user_msg = (
"Suggest 3-5 topic names for this document. "
"Return ONLY valid JSON: {\"suggested_topics\": [\"topic1\", \"topic2\"]}\n\n"
f"Document text:\n{document_text[:MAX_AI_CHARS]}"
f"Document text:\n{self._truncate(document_text)}"
)
client = self._client()
response = await client.messages.create(
response = await self._client.messages.create(
model=self._model,
max_tokens=256,
system=system_prompt,
messages=[{"role": "user", "content": user_msg}],
output_config={"format": {"type": "json_schema", "schema": _SUGGESTIONS_SCHEMA}},
)
raw = response.content[0].text
stop_reason = getattr(response, "stop_reason", "end_turn")
if response.content and stop_reason == "end_turn":
raw = response.content[0].text
else:
raw = ""
return parse_suggestions(raw)
async def health_check(self) -> bool:
"""Verify API key validity and connectivity by sending a minimal message.
Does NOT pass output_config — the response shape does not matter here;
this only confirms the api_key and network path are working.
"""
try:
client = self._client()
await client.messages.create(
await self._client.messages.create(
model=self._model,
max_tokens=5,
max_tokens=8,
messages=[{"role": "user", "content": "ping"}],
)
return True
except Exception:
return False
backend/tests/test_ai_providers.py
+77 -4
View File
@@ -5,11 +5,12 @@ Wave 2 (Plan 07-02) promotes: test_get_provider_typed, test_client_singleton,
test_generic_openai_json_mode, test_context_chars_truncation, test_smart_truncation,
test_gemini_fallback_to_parse_classification.
Remaining stubs (promoted in Plan 07-03): test_anthropic_structured_output.
Wave 3 (Plan 07-03) promotes: test_anthropic_structured_output.
"""
import pytest
from unittest.mock import AsyncMock, MagicMock, patch
from ai.anthropic_provider import AnthropicProvider, _CLASSIFICATION_SCHEMA
from ai.generic_openai_provider import GenericOpenAIProvider
from ai.openai_provider import OpenAIProvider
from ai.provider_config import ProviderConfig, PROVIDER_DEFAULTS
@@ -189,9 +190,81 @@ async def test_gemini_fallback_to_parse_classification():
# ---------------------------------------------------------------------------
# Stub: promoted in Plan 07-03
# Task 1 (Plan 07-03): AnthropicProvider output_config structured output — D-03
# ---------------------------------------------------------------------------
@pytest.mark.xfail(strict=False, reason="Wave 0 stub — promoted in Plan 07-03")
@pytest.mark.asyncio
async def test_anthropic_structured_output():
pytest.xfail("not implemented yet — Plan 07-03")
"""AnthropicProvider.classify() passes output_config with the classification schema (D-03).
Verifies:
- output_config kwarg is present in the messages.create call
- output_config value matches _CLASSIFICATION_SCHEMA exactly
- Provider correctly reads response.content[0].text when stop_reason == "end_turn"
- Singleton _client is reused (AsyncAnthropic constructed once per provider instance)
"""
provider = AnthropicProvider(api_key="test-key", model="claude-sonnet-4-6", context_chars=100)
# Build a stub response that models a successful end_turn response
stub_content = MagicMock()
stub_content.text = '{"assigned_topics":[],"new_topic_suggestions":[]}'
stub_response = MagicMock()
stub_response.content = [stub_content]
stub_response.stop_reason = "end_turn"
mock_create = AsyncMock(return_value=stub_response)
with patch("ai.anthropic_provider.anthropic.AsyncAnthropic") as mock_cls:
mock_client = MagicMock()
mock_client.messages = MagicMock()
mock_client.messages.create = mock_create
mock_cls.return_value = mock_client
# Re-create provider inside the patch so self._client uses the mock
provider = AnthropicProvider(
api_key="test-key", model="claude-sonnet-4-6", context_chars=100
)
result = await provider.classify("short doc text", [], "sys prompt")
# output_config must be present and match the schema (D-03)
call_kwargs = mock_create.await_args.kwargs
assert "output_config" in call_kwargs, "output_config must be passed to messages.create()"
assert call_kwargs["output_config"] == {
"format": {"type": "json_schema", "schema": _CLASSIFICATION_SCHEMA}
}, "output_config must use _CLASSIFICATION_SCHEMA"
# AsyncAnthropic must have been constructed exactly once (D-07 singleton)
assert mock_cls.call_count == 1, "AsyncAnthropic must be constructed once (singleton)"
# Result must be a valid ClassificationResult
assert result.topics == []
assert result.suggested_new_topics == []
@pytest.mark.asyncio
async def test_anthropic_stop_reason_fallback():
"""When stop_reason is not 'end_turn', AnthropicProvider falls back to empty ClassificationResult.
T-07-08: refusal or max_tokens stop_reason must not crash — parse_classification("") returns
an empty result rather than raising an exception.
"""
stub_content = MagicMock()
stub_content.text = '{"assigned_topics":["should","be","ignored"]}'
stub_response = MagicMock()
stub_response.content = [stub_content]
stub_response.stop_reason = "max_tokens" # simulated refusal / truncation
mock_create = AsyncMock(return_value=stub_response)
with patch("ai.anthropic_provider.anthropic.AsyncAnthropic") as mock_cls:
mock_client = MagicMock()
mock_client.messages = MagicMock()
mock_client.messages.create = mock_create
mock_cls.return_value = mock_client
provider = AnthropicProvider(api_key="k", model="claude-sonnet-4-6", context_chars=1000)
result = await provider.classify("doc text", [], "sys")
# stop_reason != "end_turn" → raw == "" → empty ClassificationResult, no crash
assert result.topics == []
assert result.suggested_new_topics == []