saksham
1a4c80958f
FastAPI + Celery + Next.js + Postgres/Redis app with company monitoring, source collection, LLM-based change analysis, enrichment, and account security (Turnstile, escalating lockout, email verification).
83 lines
3.1 KiB
Python
83 lines
3.1 KiB
Python
"""Anthropic provider: structured output via forced tool-use (the response
|
|
schema becomes the tool's input_schema, so the model can only "call" it with
|
|
arguments matching the shape we asked for), with a bounded repair loop for
|
|
the rare malformed response.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from anthropic import AsyncAnthropic
|
|
from pydantic import BaseModel, ValidationError
|
|
|
|
from app.analysis.llm.base import LLMResponseError
|
|
from app.core.config import Settings
|
|
|
|
_TOOL_NAME = "emit_result"
|
|
|
|
|
|
class AnthropicLLMProvider:
|
|
provider_name = "anthropic"
|
|
|
|
def __init__(self, settings: Settings) -> None:
|
|
self._settings = settings
|
|
self._client = AsyncAnthropic(api_key=settings.anthropic_api_key)
|
|
|
|
async def generate_structured[T: BaseModel](
|
|
self, system_prompt: str, user_prompt: str, response_model: type[T]
|
|
) -> T:
|
|
tools = [
|
|
{
|
|
"name": _TOOL_NAME,
|
|
"description": f"Emit the result matching the {response_model.__name__} schema.",
|
|
"input_schema": response_model.model_json_schema(),
|
|
}
|
|
]
|
|
|
|
last_error: Exception | None = None
|
|
messages: list[dict] = [{"role": "user", "content": user_prompt}]
|
|
|
|
for attempt in range(self._settings.llm_max_retries + 1):
|
|
if attempt > 0 and last_error is not None:
|
|
messages = [
|
|
*messages,
|
|
{
|
|
"role": "user",
|
|
"content": (
|
|
f"Your previous response did not match the required schema: "
|
|
f"{last_error}. Please try again."
|
|
),
|
|
},
|
|
]
|
|
|
|
response = await self._client.messages.create(
|
|
model=self._settings.anthropic_model,
|
|
max_tokens=self._settings.llm_max_tokens_per_request,
|
|
system=system_prompt,
|
|
tools=tools,
|
|
tool_choice={"type": "tool", "name": _TOOL_NAME},
|
|
messages=messages,
|
|
)
|
|
tool_use = next((b for b in response.content if b.type == "tool_use"), None)
|
|
if tool_use is None:
|
|
last_error = ValueError("No tool_use block in the model's response")
|
|
continue
|
|
try:
|
|
return response_model.model_validate(tool_use.input)
|
|
except ValidationError as exc:
|
|
last_error = exc
|
|
continue
|
|
|
|
raise LLMResponseError(
|
|
f"Anthropic provider failed to produce a valid {response_model.__name__} after "
|
|
f"{self._settings.llm_max_retries + 1} attempt(s): {last_error}"
|
|
)
|
|
|
|
async def generate_text(self, system_prompt: str, user_prompt: str) -> str:
|
|
response = await self._client.messages.create(
|
|
model=self._settings.anthropic_model,
|
|
max_tokens=self._settings.llm_max_tokens_per_request,
|
|
system=system_prompt,
|
|
messages=[{"role": "user", "content": user_prompt}],
|
|
)
|
|
return "\n".join(block.text for block in response.content if block.type == "text")
|