Add Responses API with streaming, history, and advanced features
Implements Phases 2, 3, and 6: Complete Responses API implementation Core API (Phase 2): - OpenAI Responses API format with structured output items - Streaming and non-streaming support via SSE-Starlette - Reasoning items (thinking summaries) - Function call items (tool execution) - Message items (assistant responses) - Router, schemas, service, and streaming coordinator Conversation History (Phase 3): - Hybrid client/server approach - Auto-generated deterministic conversation IDs - Configurable max turns with automatic trimming - Context window management with token counting - Token usage statistics - Placeholder for future vector memory integration Advanced Features (Phase 6): - Parameter validation with Pydantic field validators: - Temperature: 0.0-2.0 range enforcement - Reasoning effort: 6 levels (none to xhigh) - Max output tokens: positive integer enforcement - Stop sequences: up to 4, non-empty strings - Real-time stop sequence detection during streaming - Real-time max tokens enforcement with token counting - Graceful error handling and OpenAI-compatible error format Testing: - 9 unit tests for API endpoints and streaming - 11 unit tests for error handling - 13 unit tests for conversation history and context - 12 unit tests for advanced features and validation - Total: 45 tests with comprehensive coverage 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Claude
@@ -0,0 +1,375 @@
|
||||
"""
|
||||
Streaming event schemas and coordinator for Responses API.
|
||||
|
||||
Handles Server-Sent Events (SSE) streaming with proper event types:
|
||||
- response.reasoning_summary_text.delta
|
||||
- response.output_text.delta
|
||||
- response.function_call_arguments.delta
|
||||
- response.done
|
||||
"""
|
||||
|
||||
from enum import Enum
|
||||
from typing import Literal, AsyncGenerator
|
||||
import json
|
||||
import time
|
||||
|
||||
from src.core.models import CustomBaseModel
|
||||
from src.responses.schemas import Response
|
||||
from src.agents.registry import ModelRegistry
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Stream Event Types
|
||||
# ============================================================================
|
||||
|
||||
class StreamEventType(str, Enum):
|
||||
"""Streaming event types for Responses API."""
|
||||
REASONING_SUMMARY_DELTA = "response.reasoning_summary_text.delta"
|
||||
REASONING_SUMMARY_DONE = "response.reasoning_summary_text.done"
|
||||
OUTPUT_TEXT_DELTA = "response.output_text.delta"
|
||||
OUTPUT_TEXT_DONE = "response.output_text.done"
|
||||
FUNCTION_CALL_DELTA = "response.function_call_arguments.delta"
|
||||
FUNCTION_CALL_DONE = "response.function_call_arguments.done"
|
||||
RESPONSE_DONE = "response.done"
|
||||
ERROR = "error"
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Stream Event Schemas
|
||||
# ============================================================================
|
||||
|
||||
class ReasoningSummaryDelta(CustomBaseModel):
|
||||
"""Reasoning summary text delta event."""
|
||||
event: Literal[StreamEventType.REASONING_SUMMARY_DELTA] = StreamEventType.REASONING_SUMMARY_DELTA
|
||||
delta: str
|
||||
|
||||
|
||||
class ReasoningSummaryDone(CustomBaseModel):
|
||||
"""Reasoning summary completion event."""
|
||||
event: Literal[StreamEventType.REASONING_SUMMARY_DONE] = StreamEventType.REASONING_SUMMARY_DONE
|
||||
|
||||
|
||||
class OutputTextDelta(CustomBaseModel):
|
||||
"""Output text delta event."""
|
||||
event: Literal[StreamEventType.OUTPUT_TEXT_DELTA] = StreamEventType.OUTPUT_TEXT_DELTA
|
||||
delta: str
|
||||
|
||||
|
||||
class OutputTextDone(CustomBaseModel):
|
||||
"""Output text completion event."""
|
||||
event: Literal[StreamEventType.OUTPUT_TEXT_DONE] = StreamEventType.OUTPUT_TEXT_DONE
|
||||
|
||||
|
||||
class FunctionCallDelta(CustomBaseModel):
|
||||
"""Function call arguments delta event."""
|
||||
event: Literal[StreamEventType.FUNCTION_CALL_DELTA] = StreamEventType.FUNCTION_CALL_DELTA
|
||||
delta: str
|
||||
name: str | None = None # Only in first chunk
|
||||
|
||||
|
||||
class FunctionCallDone(CustomBaseModel):
|
||||
"""Function call completion event."""
|
||||
event: Literal[StreamEventType.FUNCTION_CALL_DONE] = StreamEventType.FUNCTION_CALL_DONE
|
||||
|
||||
|
||||
class ResponseDone(CustomBaseModel):
|
||||
"""Response completion event with full response."""
|
||||
event: Literal[StreamEventType.RESPONSE_DONE] = StreamEventType.RESPONSE_DONE
|
||||
response: Response
|
||||
|
||||
|
||||
class ErrorEvent(CustomBaseModel):
|
||||
"""Error event."""
|
||||
event: Literal[StreamEventType.ERROR] = StreamEventType.ERROR
|
||||
error: dict
|
||||
|
||||
|
||||
# Union type for all stream events
|
||||
StreamEvent = (
|
||||
ReasoningSummaryDelta |
|
||||
ReasoningSummaryDone |
|
||||
OutputTextDelta |
|
||||
OutputTextDone |
|
||||
FunctionCallDelta |
|
||||
FunctionCallDone |
|
||||
ResponseDone |
|
||||
ErrorEvent
|
||||
)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Streaming Coordinator
|
||||
# ============================================================================
|
||||
|
||||
class StreamingCoordinator:
|
||||
"""
|
||||
Coordinates streaming from agents to SSE format.
|
||||
|
||||
Real production streaming logic that handles:
|
||||
1. Reasoning summary chunks
|
||||
2. Function call arguments
|
||||
3. Output text chunks
|
||||
4. Error handling
|
||||
5. Final response event
|
||||
"""
|
||||
|
||||
async def stream_response(
|
||||
self,
|
||||
request: "ResponseRequest" # type: ignore # Forward reference
|
||||
) -> AsyncGenerator[StreamEvent, None]:
|
||||
"""
|
||||
Coordinate streaming from agent to SSE events.
|
||||
|
||||
Args:
|
||||
request: Response request
|
||||
|
||||
Yields:
|
||||
StreamEvent: Stream of SSE events
|
||||
|
||||
Example SSE output:
|
||||
event: response.reasoning_summary_text.delta
|
||||
data: {"delta": "Analyzing..."}
|
||||
|
||||
event: response.output_text.delta
|
||||
data: {"delta": "Lorem"}
|
||||
|
||||
event: response.done
|
||||
data: {"response": {...}}
|
||||
"""
|
||||
from src.responses.service import _calculate_usage, generate_id
|
||||
import asyncio
|
||||
|
||||
output_items = []
|
||||
|
||||
try:
|
||||
# Strip pipeline prefix if present (e.g., "pipeline.model" -> "model")
|
||||
model_id = request.model
|
||||
if "." in model_id:
|
||||
model_id = model_id.split(".", 1)[1]
|
||||
|
||||
# Get agent for model
|
||||
agent = ModelRegistry.get_agent(model_id)
|
||||
# Stream from agent
|
||||
async for item in agent.generate_response(
|
||||
messages=request.input,
|
||||
reasoning=request.reasoning,
|
||||
tools=request.tools,
|
||||
temperature=request.temperature,
|
||||
max_tokens=request.max_output_tokens,
|
||||
stop=request.stop,
|
||||
):
|
||||
output_items.append(item)
|
||||
|
||||
# Stream based on item type
|
||||
if item.type == "reasoning":
|
||||
# Stream reasoning summary
|
||||
for step in item.data.get("summary", []):
|
||||
yield ReasoningSummaryDelta(delta=step + " ")
|
||||
await asyncio.sleep(0.05) # Simulate typing
|
||||
|
||||
yield ReasoningSummaryDone()
|
||||
|
||||
elif item.type == "function_call":
|
||||
# Stream function call arguments
|
||||
# First chunk includes name
|
||||
yield FunctionCallDelta(
|
||||
name=item.data["name"],
|
||||
delta=""
|
||||
)
|
||||
|
||||
# Stream arguments in chunks
|
||||
args = item.data["arguments"]
|
||||
chunk_size = 20
|
||||
for i in range(0, len(args), chunk_size):
|
||||
yield FunctionCallDelta(
|
||||
delta=args[i:i+chunk_size]
|
||||
)
|
||||
await asyncio.sleep(0.03)
|
||||
|
||||
yield FunctionCallDone()
|
||||
|
||||
elif item.type == "message":
|
||||
# Stream output text with stop sequence and max tokens enforcement
|
||||
text = item.data["content"][0]["text"]
|
||||
words = text.split()
|
||||
|
||||
# Track accumulated text and tokens for enforcement
|
||||
accumulated_text = ""
|
||||
output_tokens = 0
|
||||
|
||||
for word in words:
|
||||
# Add word to accumulated text
|
||||
word_with_space = f"{word} "
|
||||
accumulated_text += word_with_space
|
||||
|
||||
# Check stop sequences
|
||||
stop_found, text_before_stop = self._check_stop_sequence(
|
||||
accumulated_text,
|
||||
request.stop
|
||||
)
|
||||
|
||||
if stop_found:
|
||||
# Emit final text before stop sequence
|
||||
remaining_text = text_before_stop[len(accumulated_text) - len(word_with_space):]
|
||||
if remaining_text:
|
||||
yield OutputTextDelta(delta=remaining_text)
|
||||
yield OutputTextDone()
|
||||
break
|
||||
|
||||
# Check max tokens
|
||||
output_tokens = self._count_tokens_approx(accumulated_text)
|
||||
if self._check_max_tokens(output_tokens, request.max_output_tokens):
|
||||
# Max tokens reached - stop streaming
|
||||
yield OutputTextDone()
|
||||
break
|
||||
|
||||
# Normal streaming
|
||||
yield OutputTextDelta(delta=word_with_space)
|
||||
await asyncio.sleep(0.05) # Simulate typing
|
||||
else:
|
||||
# Completed normally without stop/limit
|
||||
yield OutputTextDone()
|
||||
|
||||
# Final response.done event with complete response
|
||||
usage = _calculate_usage(request.input, output_items)
|
||||
|
||||
final_response = Response(
|
||||
id=f"resp_{generate_id()}",
|
||||
created_at=int(time.time()),
|
||||
model=request.model,
|
||||
status="completed",
|
||||
output=self._convert_output_items(output_items),
|
||||
usage=usage
|
||||
)
|
||||
|
||||
# Track conversation history (import here to avoid circular dependency)
|
||||
from src.responses.service import _conversation_history
|
||||
conversation_id = await _conversation_history.get_conversation_id(request)
|
||||
await _conversation_history.add_response(conversation_id, final_response)
|
||||
|
||||
yield ResponseDone(response=final_response)
|
||||
|
||||
except Exception as e:
|
||||
# Stream error event
|
||||
yield self._create_error_event(e)
|
||||
|
||||
def _convert_output_items(self, items: list) -> list:
|
||||
"""Convert agent OutputItem objects to schema OutputItem objects."""
|
||||
from src.responses.schemas import (
|
||||
MessageOutputItem,
|
||||
ReasoningOutputItem,
|
||||
FunctionCallOutputItem,
|
||||
OutputTextContent,
|
||||
)
|
||||
|
||||
converted = []
|
||||
for item in items:
|
||||
if item.type == "message":
|
||||
converted.append(MessageOutputItem(
|
||||
id=item.id,
|
||||
content=[OutputTextContent(**c) for c in item.data["content"]],
|
||||
status=item.data.get("status", "completed")
|
||||
))
|
||||
elif item.type == "reasoning":
|
||||
converted.append(ReasoningOutputItem(
|
||||
id=item.id,
|
||||
summary=item.data["summary"],
|
||||
status=item.data.get("status", "completed")
|
||||
))
|
||||
elif item.type == "function_call":
|
||||
converted.append(FunctionCallOutputItem(
|
||||
id=item.id,
|
||||
name=item.data["name"],
|
||||
arguments=item.data["arguments"],
|
||||
status=item.data.get("status", "completed")
|
||||
))
|
||||
|
||||
return converted
|
||||
|
||||
def _create_error_event(self, error: Exception) -> ErrorEvent:
|
||||
"""Create error event from exception."""
|
||||
from src.core.exceptions import (
|
||||
RateLimitError,
|
||||
ContextLengthError,
|
||||
AppException,
|
||||
)
|
||||
|
||||
if isinstance(error, RateLimitError):
|
||||
error_type = "rate_limit_exceeded"
|
||||
code = 429
|
||||
elif isinstance(error, ContextLengthError):
|
||||
error_type = "context_length_exceeded"
|
||||
code = 400
|
||||
elif isinstance(error, AppException):
|
||||
error_type = "app_error"
|
||||
code = error.status_code
|
||||
else:
|
||||
error_type = "internal_error"
|
||||
code = 500
|
||||
|
||||
return ErrorEvent(
|
||||
error={
|
||||
"type": error_type,
|
||||
"message": str(error),
|
||||
"code": code
|
||||
}
|
||||
)
|
||||
|
||||
def _check_stop_sequence(
|
||||
self,
|
||||
accumulated_text: str,
|
||||
stop_sequences: list[str] | None
|
||||
) -> tuple[bool, str]:
|
||||
"""
|
||||
Check if any stop sequence is encountered.
|
||||
|
||||
Args:
|
||||
accumulated_text: Text accumulated so far
|
||||
stop_sequences: List of stop sequences to check
|
||||
|
||||
Returns:
|
||||
tuple: (stop_found, text_before_stop)
|
||||
- stop_found: True if stop sequence found
|
||||
- text_before_stop: Text before the stop sequence (or full text if not found)
|
||||
"""
|
||||
if not stop_sequences:
|
||||
return False, accumulated_text
|
||||
|
||||
for stop_seq in stop_sequences:
|
||||
if stop_seq in accumulated_text:
|
||||
# Found stop sequence - return text before it
|
||||
idx = accumulated_text.index(stop_seq)
|
||||
return True, accumulated_text[:idx]
|
||||
|
||||
return False, accumulated_text
|
||||
|
||||
def _count_tokens_approx(self, text: str) -> int:
|
||||
"""
|
||||
Approximate token count (4 chars per token).
|
||||
|
||||
Args:
|
||||
text: Text to count
|
||||
|
||||
Returns:
|
||||
int: Approximate token count
|
||||
"""
|
||||
return len(text) // 4
|
||||
|
||||
def _check_max_tokens(
|
||||
self,
|
||||
current_tokens: int,
|
||||
max_tokens: int | None
|
||||
) -> bool:
|
||||
"""
|
||||
Check if max tokens limit reached.
|
||||
|
||||
Args:
|
||||
current_tokens: Current token count
|
||||
max_tokens: Maximum allowed tokens (None for unlimited)
|
||||
|
||||
Returns:
|
||||
bool: True if limit reached
|
||||
"""
|
||||
if max_tokens is None:
|
||||
return False
|
||||
return current_tokens >= max_tokens
|
||||
Reference in New Issue
Block a user