diff --git a/packages/capabilities/text-generation/pyproject.toml b/packages/capabilities/text-generation/pyproject.toml index 502f8dad..0db8e1bb 100644 --- a/packages/capabilities/text-generation/pyproject.toml +++ b/packages/capabilities/text-generation/pyproject.toml @@ -27,6 +27,7 @@ Issues = "https://github.com/withceleste/celeste-python/issues" [tool.uv.sources] celeste-ai = { workspace = true } +celeste-anthropic = { workspace = true } [project.entry-points."celeste.packages"] text-generation = "celeste_text_generation:register_package" diff --git a/packages/capabilities/text-generation/src/celeste_text_generation/io.py b/packages/capabilities/text-generation/src/celeste_text_generation/io.py index 2d657821..93081137 100644 --- a/packages/capabilities/text-generation/src/celeste_text_generation/io.py +++ b/packages/capabilities/text-generation/src/celeste_text_generation/io.py @@ -15,7 +15,7 @@ class TextGenerationFinishReason(FinishReason): Stores raw provider reason. Providers map their values in implementation. """ - reason: str + reason: str | None = None class TextGenerationUsage(Usage): diff --git a/packages/capabilities/text-generation/src/celeste_text_generation/providers/anthropic/client.py b/packages/capabilities/text-generation/src/celeste_text_generation/providers/anthropic/client.py index 66c62102..080cff1f 100644 --- a/packages/capabilities/text-generation/src/celeste_text_generation/providers/anthropic/client.py +++ b/packages/capabilities/text-generation/src/celeste_text_generation/providers/anthropic/client.py @@ -1,30 +1,24 @@ """Anthropic client implementation for text generation.""" -from collections.abc import AsyncIterator from typing import Any, Unpack -import httpx -from pydantic import BaseModel +from celeste_anthropic.messages.client import AnthropicMessagesClient -from celeste.mime_types import ApplicationMimeType from celeste.parameters import ParameterMapper +from celeste.types import StructuredOutput from celeste_text_generation.client import TextGenerationClient from celeste_text_generation.io import ( TextGenerationFinishReason, TextGenerationInput, TextGenerationUsage, ) -from celeste_text_generation.parameters import ( - TextGenerationParameter, - TextGenerationParameters, -) +from celeste_text_generation.parameters import TextGenerationParameters -from . import config from .parameters import ANTHROPIC_PARAMETER_MAPPERS from .streaming import AnthropicTextGenerationStream -class AnthropicTextGenerationClient(TextGenerationClient): +class AnthropicTextGenerationClient(AnthropicMessagesClient, TextGenerationClient): """Anthropic client for text generation.""" @classmethod @@ -37,31 +31,16 @@ def _init_request(self, inputs: TextGenerationInput) -> dict[str, Any]: def _parse_usage(self, response_data: dict[str, Any]) -> TextGenerationUsage: """Parse usage from response.""" - usage_data = response_data.get("usage", {}) - input_tokens = usage_data.get("input_tokens") - output_tokens = usage_data.get("output_tokens") - - total_tokens = None - if input_tokens is not None and output_tokens is not None: - total_tokens = input_tokens + output_tokens - - return TextGenerationUsage( - input_tokens=input_tokens, - output_tokens=output_tokens, - total_tokens=total_tokens, - cached_tokens=usage_data.get("cache_read_input_tokens"), - ) + usage = super()._parse_usage(response_data) + return TextGenerationUsage(**usage) def _parse_content( self, response_data: dict[str, Any], **parameters: Unpack[TextGenerationParameters], - ) -> str | BaseModel: + ) -> StructuredOutput: """Parse content from response.""" - content = response_data.get("content", []) - if not content: - msg = "No content blocks in response" - raise ValueError(msg) + content = super()._parse_content(response_data) text_content = "" for content_block in content: @@ -73,75 +52,14 @@ def _parse_content( def _parse_finish_reason( self, response_data: dict[str, Any] - ) -> TextGenerationFinishReason | None: + ) -> TextGenerationFinishReason: """Parse finish reason from response.""" - stop_reason = response_data.get("stop_reason") - if stop_reason is None: - return None - - return TextGenerationFinishReason(reason=stop_reason) - - def _build_metadata(self, response_data: dict[str, Any]) -> dict[str, Any]: - """Build metadata dictionary from response data.""" - # Filter content field before calling super - content_fields = {"content"} - filtered_data = { - k: v for k, v in response_data.items() if k not in content_fields - } - return super()._build_metadata(filtered_data) - - async def _make_request( - self, - request_body: dict[str, Any], - **parameters: Unpack[TextGenerationParameters], - ) -> httpx.Response: - """Make HTTP request(s) and return response object.""" - request_body["model"] = self.model.id - request_body["max_tokens"] = parameters.get("max_tokens") or 1024 - - headers = { - **self.auth.get_headers(), - config.ANTHROPIC_VERSION_HEADER: config.ANTHROPIC_VERSION, - "Content-Type": ApplicationMimeType.JSON, - } - - if parameters.get(TextGenerationParameter.OUTPUT_SCHEMA) is not None: - headers[config.ANTHROPIC_BETA_HEADER] = config.STRUCTURED_OUTPUTS_BETA - - return await self.http_client.post( - f"{config.BASE_URL}{config.ENDPOINT}", - headers=headers, - json_body=request_body, - ) + finish_reason = super()._parse_finish_reason(response_data) + return TextGenerationFinishReason(reason=finish_reason.reason) def _stream_class(self) -> type[AnthropicTextGenerationStream]: """Return the Stream class for this client.""" return AnthropicTextGenerationStream - def _make_stream_request( - self, - request_body: dict[str, Any], - **parameters: Unpack[TextGenerationParameters], - ) -> AsyncIterator[dict[str, Any]]: - """Make HTTP streaming request and return async iterator of events.""" - request_body["model"] = self.model.id - request_body["max_tokens"] = parameters.get("max_tokens") or 1024 - request_body["stream"] = True - - headers = { - **self.auth.get_headers(), - config.ANTHROPIC_VERSION_HEADER: config.ANTHROPIC_VERSION, - "Content-Type": ApplicationMimeType.JSON, - } - - if parameters.get(TextGenerationParameter.OUTPUT_SCHEMA) is not None: - headers[config.ANTHROPIC_BETA_HEADER] = config.STRUCTURED_OUTPUTS_BETA - - return self.http_client.stream_post( - f"{config.BASE_URL}{config.STREAM_ENDPOINT}", - headers=headers, - json_body=request_body, - ) - __all__ = ["AnthropicTextGenerationClient"] diff --git a/packages/capabilities/text-generation/src/celeste_text_generation/providers/anthropic/config.py b/packages/capabilities/text-generation/src/celeste_text_generation/providers/anthropic/config.py deleted file mode 100644 index a4437062..00000000 --- a/packages/capabilities/text-generation/src/celeste_text_generation/providers/anthropic/config.py +++ /dev/null @@ -1,18 +0,0 @@ -"""Anthropic provider configuration for text generation.""" - -# HTTP Configuration -BASE_URL = "https://api.anthropic.com" -ENDPOINT = "/v1/messages" -STREAM_ENDPOINT = ENDPOINT - -# Authentication -AUTH_HEADER_NAME = "x-api-key" -AUTH_HEADER_PREFIX = "" - -# API Version Header (required by Anthropic) -ANTHROPIC_VERSION_HEADER = "anthropic-version" -ANTHROPIC_VERSION = "2023-06-01" - -# Beta Features -ANTHROPIC_BETA_HEADER = "anthropic-beta" -STRUCTURED_OUTPUTS_BETA = "structured-outputs-2025-11-13" diff --git a/packages/capabilities/text-generation/src/celeste_text_generation/providers/anthropic/models.py b/packages/capabilities/text-generation/src/celeste_text_generation/providers/anthropic/models.py index 76d2690e..517d97b9 100644 --- a/packages/capabilities/text-generation/src/celeste_text_generation/providers/anthropic/models.py +++ b/packages/capabilities/text-generation/src/celeste_text_generation/providers/anthropic/models.py @@ -2,6 +2,7 @@ from celeste import Model, Provider from celeste.constraints import Range, Schema +from celeste.core import Parameter from celeste_text_generation.parameters import TextGenerationParameter MODELS: list[Model] = [ @@ -11,6 +12,7 @@ display_name="Claude Sonnet 4.5", streaming=True, parameter_constraints={ + Parameter.MAX_TOKENS: Range(min=1, max=64000), TextGenerationParameter.THINKING_BUDGET: Range(min=-1, max=64000), TextGenerationParameter.OUTPUT_SCHEMA: Schema(), }, @@ -21,6 +23,7 @@ display_name="Claude Haiku 4.5", streaming=True, parameter_constraints={ + Parameter.MAX_TOKENS: Range(min=1, max=64000), TextGenerationParameter.THINKING_BUDGET: Range(min=-1, max=32000), }, ), @@ -30,6 +33,18 @@ display_name="Claude Opus 4.1", streaming=True, parameter_constraints={ + Parameter.MAX_TOKENS: Range(min=1, max=32000), + TextGenerationParameter.THINKING_BUDGET: Range(min=-1, max=32000), + TextGenerationParameter.OUTPUT_SCHEMA: Schema(), + }, + ), + Model( + id="claude-opus-4-5", + provider=Provider.ANTHROPIC, + display_name="Claude Opus 4.5", + streaming=True, + parameter_constraints={ + Parameter.MAX_TOKENS: Range(min=1, max=64000), TextGenerationParameter.THINKING_BUDGET: Range(min=-1, max=32000), TextGenerationParameter.OUTPUT_SCHEMA: Schema(), }, @@ -40,6 +55,7 @@ display_name="Claude Sonnet 4", streaming=True, parameter_constraints={ + Parameter.MAX_TOKENS: Range(min=1, max=64000), TextGenerationParameter.THINKING_BUDGET: Range(min=-1, max=64000), }, ), @@ -49,6 +65,7 @@ display_name="Claude Opus 4", streaming=True, parameter_constraints={ + Parameter.MAX_TOKENS: Range(min=1, max=32000), TextGenerationParameter.THINKING_BUDGET: Range(min=-1, max=32000), }, ), diff --git a/packages/capabilities/text-generation/src/celeste_text_generation/providers/anthropic/parameters.py b/packages/capabilities/text-generation/src/celeste_text_generation/providers/anthropic/parameters.py index 80d3437f..4cd0ee9c 100644 --- a/packages/capabilities/text-generation/src/celeste_text_generation/providers/anthropic/parameters.py +++ b/packages/capabilities/text-generation/src/celeste_text_generation/providers/anthropic/parameters.py @@ -1,69 +1,38 @@ -"""Anthropic parameter mappers for text generation.""" - -import json -from typing import Any, get_args, get_origin - -from pydantic import BaseModel, TypeAdapter - -from celeste.exceptions import ConstraintViolationError +"""Anthropic Messages parameter mappers for text generation.""" + +from typing import Any + +from celeste_anthropic.messages.parameters import ( + MaxTokensMapper as _MaxTokensMapper, +) +from celeste_anthropic.messages.parameters import ( + OutputSchemaMapper as _OutputSchemaMapper, +) +from celeste_anthropic.messages.parameters import ( + TemperatureMapper as _TemperatureMapper, +) +from celeste_anthropic.messages.parameters import ( + ThinkingMapper as _ThinkingMapper, +) + +from celeste.core import Parameter from celeste.models import Model from celeste.parameters import ParameterMapper from celeste_text_generation.parameters import TextGenerationParameter -class ThinkingBudgetMapper(ParameterMapper): - """Map thinking_budget parameter to Anthropic thinking.budget_tokens.""" - - name = TextGenerationParameter.THINKING_BUDGET - - def map( - self, - request: dict[str, Any], - value: object, - model: Model, - ) -> dict[str, Any]: - """Transform thinking_budget into provider request. - - Maps unified thinking_budget to Anthropic thinking parameter: - - thinking_budget=None → No thinking parameter (thinking disabled) - - thinking_budget=-1 → {"type": "auto"} (dynamic budget, automatic) - - thinking_budget=N (where N >= 1024) → {"type": "enabled", "budget_tokens": N} (fixed budget) +class TemperatureMapper(_TemperatureMapper): + name = Parameter.TEMPERATURE - Args: - request: Provider request dict. - value: thinking_budget value (int | None). - model: Model instance containing parameter_constraints for validation. - model: Model instance with parameter constraints for validation. - Returns: - Updated request dict with thinking parameter if value provided. +class MaxTokensMapper(_MaxTokensMapper): + name = Parameter.MAX_TOKENS - Raises: - ConstraintViolationError: If value is not -1 and is less than 1024 (minimum required). - """ - validated_value = self._validate_value(value, model) - if validated_value is None: - return request - # Build thinking parameter object - if validated_value == -1: - # Dynamic thinking: use "auto" type (no budget_tokens) - thinking_config: dict[str, Any] = {"type": "auto"} - else: - # Fixed budget: validate minimum is 1024 - if validated_value < 1024: - msg = f"thinking_budget must be -1 (dynamic) or >= 1024 for {model.id}, got {validated_value}" - raise ConstraintViolationError(msg) - thinking_config = {"type": "enabled", "budget_tokens": validated_value} +class ThinkingBudgetMapper(_ThinkingMapper): + """Translate unified thinking_budget values to Anthropic-native format.""" - request["thinking"] = thinking_config - return request - - -class OutputSchemaMapper(ParameterMapper): - """Map output_schema parameter to Anthropic native structured outputs (output_format).""" - - name = TextGenerationParameter.OUTPUT_SCHEMA + name = TextGenerationParameter.THINKING_BUDGET def map( self, @@ -71,172 +40,27 @@ def map( value: object, model: Model, ) -> dict[str, Any]: - """Transform output_schema into provider request using native structured outputs. - - Converts unified output_schema to Anthropic output_format parameter: - - Uses output_format with type: "json_schema" and schema definition - - Handles both BaseModel and list[BaseModel] types - - For list[BaseModel], schema is array type directly - - Args: - request: Provider request dict. - value: output_schema value (type[BaseModel] | None). - model: Model instance containing parameter_constraints for validation. - - Returns: - Updated request dict with output_format if value provided. - """ + """Transform thinking_budget with unified value translation.""" validated_value = self._validate_value(value, model) if validated_value is None: return request - schema = self._convert_to_json_schema(validated_value) - - request["output_format"] = { - "type": "json_schema", - "schema": schema, - } - - return request - - def parse_output( - self, content: str | dict[str, Any], value: object | None - ) -> str | BaseModel: - """Parse JSON text from response to BaseModel instance. - - With native structured outputs, content is direct JSON text in content[0].text. - For list[BaseModel], content is array directly. + # Translate unified → provider-native + if validated_value == -1: + provider_value = "auto" + else: + provider_value = validated_value - Args: - content: JSON string from content[0].text. - value: Original output_schema parameter value. + return super().map(request, provider_value, model) - Returns: - BaseModel instance if value provided, otherwise str unchanged. - """ - if value is None: - return content if isinstance(content, str) else json.dumps(content) - if isinstance(content, dict): - parsed_json = content - else: - parsed_json = json.loads(content) - - return TypeAdapter(value).validate_json(json.dumps(parsed_json)) - - def _convert_to_json_schema(self, output_schema: Any) -> dict[str, Any]: # noqa: ANN401 - """Convert Pydantic BaseModel or list[BaseModel] to JSON Schema format. - - For native structured outputs, list[T] is array type directly. - Ensures all object types have additionalProperties: false as required by Anthropic. - - Args: - output_schema: Pydantic BaseModel class or list[BaseModel] type. - - Returns: - JSON Schema dictionary compatible with Anthropic structured outputs. - """ - origin = get_origin(output_schema) - if origin is list: - inner_type = get_args(output_schema)[0] - items_schema = inner_type.model_json_schema() - items_schema = self._resolve_refs(items_schema) - json_schema = { - "type": "array", - "items": items_schema, - } - else: - json_schema = output_schema.model_json_schema() - json_schema = self._resolve_refs(json_schema) - - json_schema = self._ensure_additional_properties(json_schema) - return json_schema - - def _ensure_additional_properties(self, schema: dict[str, Any]) -> dict[str, Any]: - """Ensure all object types have additionalProperties: false.""" - if not isinstance(schema, dict): - return schema - - schema = schema.copy() - - if schema.get("type") == "object": - schema["additionalProperties"] = False - - for key in ["properties", "items"]: - if key in schema: - if key == "properties": - schema[key] = { - k: self._ensure_additional_properties(v) - for k, v in schema[key].items() - } - else: - schema[key] = self._ensure_additional_properties(schema[key]) - - for key in ["anyOf", "allOf"]: - if key in schema: - schema[key] = [ - self._ensure_additional_properties(item) for item in schema[key] - ] - - return schema - - def _resolve_refs(self, schema: dict[str, Any]) -> dict[str, Any]: - """Resolve all $ref references and inline definitions. - - Args: - schema: JSON Schema dictionary potentially containing $ref. - - Returns: - Schema with $ref references resolved inline. - """ - defs: dict[str, Any] = {} - - def collect_defs(value: Any) -> None: # noqa: ANN401 - """Recursively collect all $defs dictionaries.""" - if isinstance(value, dict): - if "$defs" in value: - defs.update(value["$defs"]) - for v in value.values(): - collect_defs(v) - elif isinstance(value, list): - for item in value: - collect_defs(item) - - collect_defs(schema) - - def remove_defs(value: Any) -> Any: # noqa: ANN401 - """Recursively remove all $defs keys.""" - if isinstance(value, dict): - result = {k: remove_defs(v) for k, v in value.items() if k != "$defs"} - return result - elif isinstance(value, list): - return [remove_defs(item) for item in value] - return value - - schema = remove_defs(schema) - - def resolve(value: Any) -> Any: # noqa: ANN401 - """Recursively resolve $ref references in schema.""" - if isinstance(value, dict): - if "$ref" in value: - ref_path = value["$ref"] - if ref_path.startswith("#/$defs/"): - ref_name = ref_path.split("/")[-1] - if ref_name in defs: - resolved = defs[ref_name].copy() - resolved.update( - {k: v for k, v in value.items() if k != "$ref"} - ) - return resolve(resolved) - return {k: resolve(v) for k, v in value.items()} - elif isinstance(value, list): - return [resolve(item) for item in value] - return value - - return resolve(schema) +class OutputSchemaMapper(_OutputSchemaMapper): + name = TextGenerationParameter.OUTPUT_SCHEMA ANTHROPIC_PARAMETER_MAPPERS: list[ParameterMapper] = [ + TemperatureMapper(), + MaxTokensMapper(), ThinkingBudgetMapper(), OutputSchemaMapper(), ] diff --git a/packages/capabilities/text-generation/src/celeste_text_generation/providers/anthropic/streaming.py b/packages/capabilities/text-generation/src/celeste_text_generation/providers/anthropic/streaming.py index 90c22af2..bfcc8e4b 100644 --- a/packages/capabilities/text-generation/src/celeste_text_generation/providers/anthropic/streaming.py +++ b/packages/capabilities/text-generation/src/celeste_text_generation/providers/anthropic/streaming.py @@ -3,6 +3,9 @@ from collections.abc import Callable from typing import Any, Unpack +from celeste_anthropic.messages.streaming import AnthropicMessagesStream + +from celeste.types import StructuredOutput from celeste_text_generation.io import ( TextGenerationChunk, TextGenerationFinishReason, @@ -13,112 +16,53 @@ from celeste_text_generation.streaming import TextGenerationStream -class AnthropicTextGenerationStream(TextGenerationStream): +class AnthropicTextGenerationStream(AnthropicMessagesStream, TextGenerationStream): """Anthropic streaming for text generation.""" def __init__( self, sse_iterator: Any, # noqa: ANN401 - transform_output: Callable[..., object], + transform_output: Callable[..., StructuredOutput], **parameters: Unpack[TextGenerationParameters], ) -> None: - """Initialize stream with output transformation support. - - Args: - sse_iterator: Server-Sent Events iterator. - transform_output: Function to transform accumulated content (e.g., JSON → BaseModel). - **parameters: Parameters passed to stream() for output transformation. - """ + """Initialize stream with output transformation support.""" super().__init__(sse_iterator, **parameters) self._transform_output = transform_output self._last_finish_reason: TextGenerationFinishReason | None = None def _parse_chunk(self, event: dict[str, Any]) -> TextGenerationChunk | None: - """Parse SSE event into Chunk.""" - event_type = event.get("type") - if not event_type: + """Parse SSE event into typed Chunk.""" + raw = super()._parse_chunk(event) + if not raw: return None - if event_type == "content_block_delta": - delta = event.get("delta", {}) - if delta.get("type") == "text_delta": - text_delta = delta.get("text") - if text_delta is not None: - return TextGenerationChunk( - content=text_delta, - finish_reason=None, - usage=None, - ) - - if event_type == "message_delta": - delta = event.get("delta", {}) - stop_reason = delta.get("stop_reason") - - finish_reason: TextGenerationFinishReason | None = None - if stop_reason is not None: - finish_reason = TextGenerationFinishReason(reason=stop_reason) - self._last_finish_reason = finish_reason - - usage = self._parse_usage_from_event(event) - - return TextGenerationChunk( - content="", - finish_reason=finish_reason, - usage=usage, - ) - - # Parse message stop event (final event) - if event_type == "message_stop": - usage = self._parse_usage_from_event(event) - - return TextGenerationChunk( - content="", - finish_reason=self._last_finish_reason, - usage=usage, - ) - - # Ignore other event types - return None - - def _parse_usage_from_event( - self, event: dict[str, Any] - ) -> TextGenerationUsage | None: - """Parse usage from SSE event data. - - Args: - event: SSE event dictionary containing usage data. - - Returns: - TextGenerationUsage object if usage data present, None otherwise. - """ - usage_data = event.get("usage") - if not usage_data: - return None + usage = TextGenerationUsage(**raw["usage"]) if raw["usage"] else None + finish_reason = ( + TextGenerationFinishReason(reason=raw["finish_reason"]) + if raw["finish_reason"] + else None + ) + + if finish_reason: + self._last_finish_reason = finish_reason + + # For message_stop events, use stored finish_reason + event_type = raw["raw_event"].get("type") + if event_type == "message_stop" and self._last_finish_reason: + finish_reason = self._last_finish_reason - input_tokens = usage_data.get("input_tokens") - output_tokens = usage_data.get("output_tokens") - total_tokens = None - if input_tokens is not None and output_tokens is not None: - total_tokens = input_tokens + output_tokens - - return TextGenerationUsage( - input_tokens=input_tokens, - output_tokens=output_tokens, - total_tokens=total_tokens, - cached_tokens=usage_data.get("cache_read_input_tokens"), + return TextGenerationChunk( + content=raw["content"], + finish_reason=finish_reason, + usage=usage, + metadata={"raw_event": raw["raw_event"]}, ) def _parse_usage(self, chunks: list[TextGenerationChunk]) -> TextGenerationUsage: - """Parse usage from chunks.""" - if not chunks: - return TextGenerationUsage() - - # Usage typically appears in message_delta or message_stop events - # Search backwards for the most recent usage + """Extract usage from final chunk.""" for chunk in reversed(chunks): if chunk.usage: return chunk.usage - return TextGenerationUsage() def _parse_output( @@ -126,18 +70,25 @@ def _parse_output( chunks: list[TextGenerationChunk], **parameters: Unpack[TextGenerationParameters], ) -> TextGenerationOutput: - """Assemble chunks into final output with structured output support.""" + """Assemble chunks into final output.""" content = "".join(chunk.content for chunk in chunks) content = self._transform_output(content, **parameters) usage = self._parse_usage(chunks) finish_reason = chunks[-1].finish_reason if chunks else None + raw_events = [ + c.metadata["raw_event"] + for c in chunks + if c.metadata.get("raw_event", {}).get("type") + in ("message_delta", "message_stop") + ] + return TextGenerationOutput( content=content, usage=usage, finish_reason=finish_reason, - metadata={}, + metadata={"raw_response": raw_events}, ) diff --git a/packages/providers/anthropic/src/celeste_anthropic/messages/client.py b/packages/providers/anthropic/src/celeste_anthropic/messages/client.py index 5fb1189e..28199c32 100644 --- a/packages/providers/anthropic/src/celeste_anthropic/messages/client.py +++ b/packages/providers/anthropic/src/celeste_anthropic/messages/client.py @@ -133,17 +133,12 @@ def _parse_content(self, response_data: dict[str, Any]) -> Any: raise ValueError(msg) return content - def _parse_finish_reason( - self, response_data: dict[str, Any] - ) -> FinishReason | None: + def _parse_finish_reason(self, response_data: dict[str, Any]) -> FinishReason: """Extract finish reason from Messages API response. Returns FinishReason that capability clients wrap in their specific type. """ stop_reason = response_data.get("stop_reason") - if stop_reason is None: - return None - return FinishReason(reason=stop_reason) def _build_metadata(self, response_data: dict[str, Any]) -> dict[str, Any]: