diff --git a/IMPLEMENTATION_COMPLETE.md b/IMPLEMENTATION_COMPLETE.md deleted file mode 100644 index f90f990851..0000000000 --- a/IMPLEMENTATION_COMPLETE.md +++ /dev/null @@ -1,414 +0,0 @@ -# ✅ Implementation Complete: OpenAI Response Format for Polling Via Cache - -## Summary - -Successfully updated the LiteLLM polling via cache feature to follow the official **OpenAI Response object format** as specified in: -- https://platform.openai.com/docs/api-reference/responses/object -- https://platform.openai.com/docs/api-reference/responses-streaming - -## What Was Implemented - -### 1. ✅ Response Object Format (OpenAI Compatible) - -The cached response object now follows OpenAI's exact structure: - -```json -{ - "id": "litellm_poll_abc123", - "object": "response", - "status": "in_progress" | "completed" | "cancelled" | "failed", - "status_details": { - "type": "completed", - "reason": "stop", - "error": {...} - }, - "output": [ - { - "id": "item_001", - "type": "message", - "content": [{"type": "text", "text": "..."}] - } - ], - "usage": { - "input_tokens": 100, - "output_tokens": 500, - "total_tokens": 600 - }, - "metadata": {...}, - "created_at": 1700000000 -} -``` - -### 2. ✅ Streaming Events Processing - -The background task now processes OpenAI's streaming events: -- `response.output_item.added` - New output items -- `response.content_part.added` - Incremental content updates -- `response.content_part.done` - Completed content parts -- `response.output_item.done` - Completed output items -- `response.done` - Final response with usage - -### 3. ✅ Redis Cache Storage - -Response objects are stored in Redis following OpenAI format: -- **Key**: `litellm:polling:response:litellm_poll_{uuid}` -- **Value**: Complete OpenAI Response object (JSON) -- **TTL**: Configurable (default: 3600s) -- **Internal State**: Tracked in `_polling_state` field - -### 4. ✅ Status Values Aligned - -| LiteLLM Status | OpenAI Status | -|---------------|---------------| -| ~~pending~~ | `in_progress` | -| ~~streaming~~ | `in_progress` | -| `completed` | `completed` | -| ~~error~~ | `failed` | -| `cancelled` | `cancelled` | - -### 5. ✅ Structured Output Items - -Content is now returned as structured output items: -- **Type**: `message`, `function_call`, `function_call_output` -- **Content**: Array of content parts (text, audio, etc.) -- **Status**: Per-item status tracking -- **ID**: Unique identifier for each output item - -### 6. ✅ Usage Tracking - -Token usage is now captured and returned: -```json -{ - "usage": { - "input_tokens": 100, - "output_tokens": 500, - "total_tokens": 600 - } -} -``` - -### 7. ✅ Enhanced Error Handling - -Errors now follow OpenAI's structured format: -```json -{ - "status": "failed", - "status_details": { - "type": "failed", - "error": { - "type": "internal_error", - "message": "Detailed error message", - "code": "error_code" - } - } -} -``` - -## Files Modified - -### Core Implementation - -1. **`litellm/proxy/response_polling/polling_handler.py`** - - ✅ Updated `create_initial_state()` to create OpenAI format - - ✅ Updated `update_state()` to handle output items and usage - - ✅ Updated `cancel_polling()` to set proper status_details - - ✅ Fixed UUID generation (using `uuid4()`) - - ✅ No linting errors - -2. **`litellm/proxy/response_api_endpoints/endpoints.py`** - - ✅ Updated `_background_streaming_task()` to process OpenAI events - - ✅ Updated POST endpoint to return OpenAI format response - - ✅ Updated GET endpoint to return OpenAI format response - - ✅ No linting errors - -3. **`litellm_config.yaml`** - - ✅ Already configured with `polling_via_cache: true` - - ✅ TTL set to 7200 seconds - - ✅ No changes needed - -### Documentation Created - -4. **`OPENAI_RESPONSE_FORMAT.md`** (NEW) - - Complete format specification - - API examples and usage - - Client implementation examples - - Redis cache structure - - 400+ lines of comprehensive docs - -5. **`OPENAI_FORMAT_CHANGES_SUMMARY.md`** (NEW) - - Summary of all changes - - Before/After comparisons - - Field mappings - - Breaking changes list - - Benefits and validation checklist - -6. **`MIGRATION_GUIDE_OPENAI_FORMAT.md`** (NEW) - - Step-by-step migration guide - - Code examples (Python & TypeScript) - - Common pitfalls - - Testing checklist - - Helper functions - -7. **`IMPLEMENTATION_COMPLETE.md`** (NEW - this file) - - Implementation summary - - Testing instructions - - Quick start guide - -### Testing - -8. **`test_polling_feature.py`** (UPDATED) - - ✅ Updated to validate OpenAI format - - ✅ Helper function to extract text content - - ✅ Tests output items, usage, status_details - - ✅ Comprehensive test coverage - -## How to Test - -### 1. Start Redis (if not running) - -```bash -redis-server -``` - -### 2. Start LiteLLM Proxy - -```bash -cd /Users/xianzongxie/stripe/litellm -litellm --config litellm_config.yaml -``` - -### 3. Run Tests - -```bash -python test_polling_feature.py -``` - -### 4. Manual Test - -```bash -# Start a background response -curl -X POST http://localhost:4000/v1/responses \ - -H "Authorization: Bearer sk-test-key" \ - -H "Content-Type: application/json" \ - -d '{ - "model": "gpt-4o", - "input": "Write a short poem", - "background": true, - "metadata": {"test": "manual"} - }' - -# Save the returned ID and poll for updates -curl -X GET http://localhost:4000/v1/responses/litellm_poll_XXXXX \ - -H "Authorization: Bearer sk-test-key" -``` - -## API Usage Examples - -### Python Client - -```python -import requests -import time - -def extract_text_content(response_obj): - """Extract text from OpenAI Response object""" - text = "" - for item in response_obj.get("output", []): - if item.get("type") == "message": - for part in item.get("content", []): - if part.get("type") == "text": - text += part.get("text", "") - return text - -# Create background response -response = requests.post( - "http://localhost:4000/v1/responses", - headers={"Authorization": "Bearer sk-test-key"}, - json={ - "model": "gpt-4o", - "input": "Explain quantum computing", - "background": True - } -) - -polling_id = response.json()["id"] -print(f"Polling ID: {polling_id}") - -# Poll for completion -while True: - response = requests.get( - f"http://localhost:4000/v1/responses/{polling_id}", - headers={"Authorization": "Bearer sk-test-key"} - ) - - data = response.json() - status = data["status"] - content = extract_text_content(data) - - print(f"Status: {status}, Content: {len(content)} chars") - - if status == "completed": - usage = data.get("usage", {}) - print(f"✅ Done! Tokens: {usage.get('total_tokens')}") - print(f"Content: {content}") - break - elif status == "failed": - error = data.get("status_details", {}).get("error", {}) - print(f"❌ Error: {error.get('message')}") - break - - time.sleep(2) -``` - -### TypeScript Client - -```typescript -interface OpenAIResponse { - id: string; - object: "response"; - status: "in_progress" | "completed" | "failed" | "cancelled"; - output: Array<{ - type: "message"; - content?: Array<{type: "text"; text: string}>; - }>; - usage: {total_tokens: number} | null; -} - -async function pollResponse(id: string): Promise { - while (true) { - const response = await fetch(`http://localhost:4000/v1/responses/${id}`, { - headers: {Authorization: "Bearer sk-test-key"} - }); - - const data: OpenAIResponse = await response.json(); - - if (data.status === "completed") { - // Extract text - const text = data.output - .filter(item => item.type === "message") - .flatMap(item => item.content || []) - .filter(part => part.type === "text") - .map(part => part.text) - .join(""); - - return text; - } else if (data.status === "failed") { - throw new Error("Response failed"); - } - - await new Promise(resolve => setTimeout(resolve, 2000)); - } -} -``` - -## Validation Checklist - -- ✅ Response object follows OpenAI format exactly -- ✅ All streaming events are processed correctly -- ✅ Status values match OpenAI specification -- ✅ Error format is structured per OpenAI spec -- ✅ Output items support multiple types (message, function_call, etc.) -- ✅ Usage data is captured and returned -- ✅ Metadata is preserved throughout lifecycle -- ✅ Redis cache stores complete Response object -- ✅ Test script validates new format -- ✅ No linting errors in implementation -- ✅ Documentation is comprehensive -- ✅ Migration guide is available -- ✅ Helper functions provided for content extraction - -## Benefits of This Implementation - -1. **🔄 OpenAI Compatibility**: Fully compatible with OpenAI's Response API -2. **📊 Structured Data**: Rich output format with multiple content types -3. **💰 Token Tracking**: Built-in usage monitoring -4. **🔍 Better Errors**: Detailed error information with types and codes -5. **⚡ Streaming Support**: Aligned with OpenAI's streaming event format -6. **🎯 Type Safety**: Clear structure for TypeScript/typed clients -7. **📈 Scalability**: Efficient Redis caching with TTL -8. **🛠️ Extensibility**: Easy to add new output types (function calls, etc.) - -## Next Steps - -### For Development - -1. **Test with Multiple Providers** - - Test with OpenAI, Anthropic, Azure, etc. - - Verify streaming events work across providers - - Validate usage tracking for all providers - -2. **Function Calling Support** - - Test with function calling responses - - Verify `function_call` and `function_call_output` items - - Validate structured output - -3. **Performance Testing** - - Load test with multiple concurrent requests - - Monitor Redis memory usage - - Optimize cache TTL settings - -4. **Error Scenarios** - - Test provider timeouts - - Test network failures - - Test rate limit errors - -### For Production - -1. **Monitoring** - - Set up Redis monitoring - - Track polling request metrics - - Monitor cache hit/miss rates - - Alert on high memory usage - -2. **Configuration** - - Adjust TTL based on usage patterns - - Configure Redis eviction policies - - Set up Redis persistence if needed - -3. **Documentation** - - Update API documentation - - Publish migration guide - - Create client library examples - -4. **Client Updates** - - Update any existing client libraries - - Provide migration tools if needed - - Communicate breaking changes - -## Support Resources - -- **Complete Format Docs**: `OPENAI_RESPONSE_FORMAT.md` -- **Migration Guide**: `MIGRATION_GUIDE_OPENAI_FORMAT.md` -- **Changes Summary**: `OPENAI_FORMAT_CHANGES_SUMMARY.md` -- **Test Script**: `test_polling_feature.py` -- **OpenAI Docs**: https://platform.openai.com/docs/api-reference/responses - -## Success Criteria ✅ - -All success criteria have been met: - -- ✅ Response objects follow OpenAI format exactly -- ✅ Streaming events are processed correctly -- ✅ Output items are structured properly -- ✅ Usage tracking is implemented -- ✅ Status values match OpenAI spec -- ✅ Error handling is structured -- ✅ Redis caching works correctly -- ✅ Code has no linting errors -- ✅ Tests validate new format -- ✅ Documentation is comprehensive -- ✅ Migration guide is available -- ✅ Helper functions are provided - -## 🎉 Implementation Status: COMPLETE - -The polling via cache feature now fully supports the OpenAI Response object format with proper streaming event processing and Redis cache storage. - -**Ready for testing and deployment!** - ---- - -*Implementation completed on: 2024-11-19* -*Format version: OpenAI Response API v1* -*LiteLLM compatibility: v1.0+* - diff --git a/MIGRATION_GUIDE_OPENAI_FORMAT.md b/MIGRATION_GUIDE_OPENAI_FORMAT.md deleted file mode 100644 index 99d26778b9..0000000000 --- a/MIGRATION_GUIDE_OPENAI_FORMAT.md +++ /dev/null @@ -1,541 +0,0 @@ -# Migration Guide: OpenAI Response Format - -This guide helps you migrate from the previous polling format to the new OpenAI Response object format. - -## Quick Reference - -### Field Name Changes - -| Old Field | New Field | Location | Notes | -|-----------|-----------|----------|-------| -| `polling_id` | `id` | Top level | Renamed for OpenAI compatibility | -| `object: "response.polling"` | `object: "response"` | Top level | Changed to match OpenAI | -| `content` (string) | `output[].content[]` | Nested | Now structured array | -| `chunks` | N/A | Removed | Data now in `output` items | -| `error` (string) | `status_details.error` (object) | Nested | Structured error format | -| `final_response` | N/A | Removed | Full data always in response | -| `content_length` | N/A | Removed | Calculate from `output` | -| `chunk_count` | N/A | Removed | Use `output.length` | - -### Status Value Changes - -| Old Status | New Status | -|-----------|-----------| -| `pending` | `in_progress` | -| `streaming` | `in_progress` | -| `completed` | `completed` | -| `error` | `failed` | -| `cancelled` | `cancelled` | - -## Code Migration Examples - -### 1. Extracting Text Content - -**Before:** -```python -response = requests.get(f"{url}/v1/responses/{polling_id}") -data = response.json() - -content = data.get("content", "") -content_length = data.get("content_length", 0) -``` - -**After:** -```python -response = requests.get(f"{url}/v1/responses/{polling_id}") -data = response.json() - -# Extract text from output items -content = "" -for item in data.get("output", []): - if item.get("type") == "message": - for part in item.get("content", []): - if part.get("type") == "text": - content += part.get("text", "") - -content_length = len(content) -``` - -**Helper Function:** -```python -def extract_text_content(response_obj): - """Extract text content from OpenAI Response object""" - text = "" - for item in response_obj.get("output", []): - if item.get("type") == "message": - for part in item.get("content", []): - if part.get("type") == "text": - text += part.get("text", "") - return text - -# Usage -content = extract_text_content(data) -``` - -### 2. Checking Status - -**Before:** -```python -status = data.get("status") - -if status == "pending" or status == "streaming": - print("Still processing...") -elif status == "completed": - print("Done!") -elif status == "error": - error_msg = data.get("error", "Unknown error") - print(f"Error: {error_msg}") -``` - -**After:** -```python -status = data.get("status") - -if status == "in_progress": - print("Still processing...") -elif status == "completed": - print("Done!") - # Check completion details - status_details = data.get("status_details", {}) - reason = status_details.get("reason", "unknown") - print(f"Completed: {reason}") -elif status == "failed": - # Structured error object - error = data.get("status_details", {}).get("error", {}) - error_type = error.get("type", "unknown") - error_msg = error.get("message", "Unknown error") - error_code = error.get("code", "") - print(f"Error [{error_type}]: {error_msg} (code: {error_code})") -``` - -### 3. Polling Loop - -**Before:** -```python -while True: - response = requests.get(f"{url}/v1/responses/{polling_id}") - data = response.json() - - status = data["status"] - content = data.get("content", "") - - print(f"Status: {status}, Content: {len(content)} chars") - - if status == "completed": - return data - elif status == "error": - raise Exception(data.get("error")) - - time.sleep(2) -``` - -**After:** -```python -def extract_text_content(response_obj): - text = "" - for item in response_obj.get("output", []): - if item.get("type") == "message": - for part in item.get("content", []): - if part.get("type") == "text": - text += part.get("text", "") - return text - -while True: - response = requests.get(f"{url}/v1/responses/{polling_id}") - data = response.json() - - status = data["status"] - content = extract_text_content(data) - - print(f"Status: {status}, Content: {len(content)} chars") - - if status == "completed": - # Show usage if available - usage = data.get("usage") - if usage: - print(f"Tokens used: {usage.get('total_tokens')}") - return data - elif status == "failed": - error = data.get("status_details", {}).get("error", {}) - raise Exception(error.get("message", "Unknown error")) - elif status == "cancelled": - raise Exception("Response was cancelled") - - time.sleep(2) -``` - -### 4. Creating Background Response - -**Before & After (Same):** -```python -response = requests.post( - f"{url}/v1/responses", - headers={"Authorization": f"Bearer {api_key}"}, - json={ - "model": "gpt-4o", - "input": "Your prompt", - "background": True - } -) - -data = response.json() -polling_id = data["id"] # Still works! (was polling_id, now just id) -``` - -**Note:** The request format is unchanged, but the response structure is different. - -### 5. Error Handling - -**Before:** -```python -if data.get("status") == "error": - error_message = data.get("error", "Unknown error") - print(f"Error: {error_message}") -``` - -**After:** -```python -if data.get("status") == "failed": - status_details = data.get("status_details", {}) - error = status_details.get("error", {}) - - error_type = error.get("type", "unknown") - error_message = error.get("message", "Unknown error") - error_code = error.get("code", "") - - print(f"Error [{error_type}]: {error_message}") - if error_code: - print(f"Error code: {error_code}") -``` - -### 6. Accessing Metadata - -**Before & After (Similar):** -```python -metadata = data.get("metadata", {}) -``` - -**Note:** Metadata structure is unchanged. - -### 7. Getting Usage Information - -**Before:** -```python -# Not available in old format -``` - -**After:** -```python -usage = data.get("usage") -if usage: - input_tokens = usage.get("input_tokens", 0) - output_tokens = usage.get("output_tokens", 0) - total_tokens = usage.get("total_tokens", 0) - - print(f"Token usage:") - print(f" Input: {input_tokens}") - print(f" Output: {output_tokens}") - print(f" Total: {total_tokens}") -``` - -## Complete Migration Example - -### Before (Old Format) - -```python -import time -import requests - -def poll_response_old(url, api_key, polling_id): - """Old format polling""" - headers = {"Authorization": f"Bearer {api_key}"} - - while True: - response = requests.get( - f"{url}/v1/responses/{polling_id}", - headers=headers - ) - data = response.json() - - status = data.get("status") - content = data.get("content", "") - content_length = data.get("content_length", 0) - - print(f"[{status}] {content_length} chars") - - if status == "completed": - print(f"✅ Done! Content: {content[:100]}...") - return content - elif status == "error": - raise Exception(f"Error: {data.get('error')}") - elif status in ["pending", "streaming"]: - time.sleep(2) - else: - raise Exception(f"Unknown status: {status}") -``` - -### After (OpenAI Format) - -```python -import time -import requests - -def extract_text_content(response_obj): - """Extract text content from OpenAI Response object""" - text = "" - for item in response_obj.get("output", []): - if item.get("type") == "message": - for part in item.get("content", []): - if part.get("type") == "text": - text += part.get("text", "") - return text - -def poll_response_new(url, api_key, polling_id): - """New OpenAI format polling""" - headers = {"Authorization": f"Bearer {api_key}"} - - while True: - response = requests.get( - f"{url}/v1/responses/{polling_id}", - headers=headers - ) - data = response.json() - - status = data.get("status") - content = extract_text_content(data) - content_length = len(content) - - print(f"[{status}] {content_length} chars") - - if status == "completed": - usage = data.get("usage", {}) - tokens = usage.get("total_tokens", 0) - print(f"✅ Done! Content: {content[:100]}...") - print(f"Tokens used: {tokens}") - return content - elif status == "failed": - error = data.get("status_details", {}).get("error", {}) - raise Exception(f"Error: {error.get('message', 'Unknown error')}") - elif status == "cancelled": - raise Exception("Response was cancelled") - elif status == "in_progress": - time.sleep(2) - else: - raise Exception(f"Unknown status: {status}") -``` - -## TypeScript/JavaScript Migration - -### Before - -```typescript -interface OldPollingResponse { - polling_id: string; - object: "response.polling"; - status: "pending" | "streaming" | "completed" | "error" | "cancelled"; - content: string; - content_length: number; - chunk_count: number; - error?: string; - metadata?: Record; -} - -// Usage -const data: OldPollingResponse = await response.json(); -console.log(data.content); -``` - -### After - -```typescript -interface OpenAIResponseObject { - id: string; - object: "response"; - status: "in_progress" | "completed" | "cancelled" | "failed" | "incomplete"; - status_details: { - type: string; - reason?: string; - error?: { - type: string; - message: string; - code: string; - }; - } | null; - output: Array<{ - id: string; - type: "message" | "function_call" | "function_call_output"; - role?: "assistant"; - status?: "in_progress" | "completed"; - content?: Array<{ - type: "text"; - text: string; - }>; - }>; - usage: { - input_tokens: number; - output_tokens: number; - total_tokens: number; - } | null; - metadata: Record; - created_at: number; -} - -// Helper function -function extractTextContent(response: OpenAIResponseObject): string { - let text = ""; - for (const item of response.output) { - if (item.type === "message" && item.content) { - for (const part of item.content) { - if (part.type === "text") { - text += part.text; - } - } - } - } - return text; -} - -// Usage -const data: OpenAIResponseObject = await response.json(); -const content = extractTextContent(data); -console.log(content); -``` - -## Configuration Changes - -### litellm_config.yaml - -**No changes required!** The configuration format remains the same: - -```yaml -litellm_settings: - cache: true - cache_params: - type: redis - host: "127.0.0.1" - port: "6379" - responses: - background_mode: - polling_via_cache: true - polling_ttl: 7200 -``` - -## Validation Checklist - -Use this checklist to ensure your migration is complete: - -- [ ] Updated field names (`polling_id` → `id`) -- [ ] Updated status checks (`pending`/`streaming` → `in_progress`) -- [ ] Updated error handling (`error` → `status_details.error`) -- [ ] Implemented content extraction from `output` array -- [ ] Added usage tracking (optional but recommended) -- [ ] Updated TypeScript interfaces (if applicable) -- [ ] Tested with actual API calls -- [ ] Updated documentation/comments in code -- [ ] Verified backward compatibility isn't assumed - -## Common Pitfalls - -### 1. Assuming Flat Content - -❌ **Wrong:** -```python -content = data.get("content", "") # This field no longer exists! -``` - -✅ **Correct:** -```python -content = extract_text_content(data) -``` - -### 2. Old Status Values - -❌ **Wrong:** -```python -if status == "pending" or status == "streaming": - # Will never match! -``` - -✅ **Correct:** -```python -if status == "in_progress": - # Correct! -``` - -### 3. Simple Error Messages - -❌ **Wrong:** -```python -error = data.get("error") # No longer exists at top level -``` - -✅ **Correct:** -```python -error = data.get("status_details", {}).get("error", {}).get("message") -``` - -### 4. Ignoring Output Item Types - -❌ **Wrong:** -```python -# Assuming all output is text -for item in data["output"]: - text = item["content"] # Might not be text! -``` - -✅ **Correct:** -```python -for item in data["output"]: - if item.get("type") == "message": - for part in item.get("content", []): - if part.get("type") == "text": - text = part.get("text", "") -``` - -## Testing Your Migration - -Use this simple test to verify your migration: - -```python -import requests - -url = "http://localhost:4000" -api_key = "sk-test-key" - -# Start background response -response = requests.post( - f"{url}/v1/responses", - headers={"Authorization": f"Bearer {api_key}"}, - json={ - "model": "gpt-4o", - "input": "Say hello", - "background": True - } -) - -data = response.json() - -# Verify new format -assert "id" in data, "Missing 'id' field" -assert data["object"] == "response", f"Wrong object type: {data['object']}" -assert data["status"] == "in_progress", f"Wrong initial status: {data['status']}" -assert "output" in data, "Missing 'output' field" -assert isinstance(data["output"], list), "output should be a list" - -print("✅ Migration successful! Your code is using the new format.") -``` - -## Getting Help - -- **Documentation**: See `OPENAI_RESPONSE_FORMAT.md` for complete format specification -- **Examples**: Check `test_polling_feature.py` for working examples -- **OpenAI Docs**: https://platform.openai.com/docs/api-reference/responses/object - -## Timeline - -- **Old Format**: Deprecated -- **New Format**: Current (OpenAI compatible) -- **Breaking Change**: Yes - requires code updates - -We recommend migrating as soon as possible to ensure compatibility with future updates. - diff --git a/OPENAI_FORMAT_CHANGES_SUMMARY.md b/OPENAI_FORMAT_CHANGES_SUMMARY.md deleted file mode 100644 index 1809342989..0000000000 --- a/OPENAI_FORMAT_CHANGES_SUMMARY.md +++ /dev/null @@ -1,337 +0,0 @@ -# OpenAI Response Format Implementation - Changes Summary - -This document summarizes all changes made to implement OpenAI Response object format for the polling via cache feature. - -## References - -- **OpenAI Response Object**: https://platform.openai.com/docs/api-reference/responses/object -- **OpenAI Streaming Events**: https://platform.openai.com/docs/api-reference/responses-streaming - -## Key Changes - -### 1. Response Object Structure - -**Before:** -```json -{ - "polling_id": "litellm_poll_abc123", - "object": "response.polling", - "status": "pending" | "streaming" | "completed" | "error" | "cancelled", - "content": "cumulative text content...", - "chunks": [...], - "error": "error message", - "final_response": {...} -} -``` - -**After (OpenAI Format):** -```json -{ - "id": "litellm_poll_abc123", - "object": "response", - "status": "in_progress" | "completed" | "cancelled" | "failed" | "incomplete", - "status_details": { - "type": "completed" | "cancelled" | "failed", - "reason": "stop" | "user_requested", - "error": { - "type": "internal_error", - "message": "error message", - "code": "error_code" - } - }, - "output": [ - { - "id": "item_001", - "type": "message", - "status": "completed", - "role": "assistant", - "content": [ - { - "type": "text", - "text": "Response text..." - } - ] - } - ], - "usage": { - "input_tokens": 100, - "output_tokens": 500, - "total_tokens": 600 - }, - "metadata": {...}, - "created_at": 1700000000 -} -``` - -### 2. Status Values Mapping - -| Old Status | New Status | Notes | -|------------|-----------|-------| -| `pending` | `in_progress` | Aligned with OpenAI | -| `streaming` | `in_progress` | Same as above | -| `completed` | `completed` | No change | -| `error` | `failed` | OpenAI format | -| `cancelled` | `cancelled` | No change | - -### 3. File Changes - -#### A. `litellm/proxy/response_polling/polling_handler.py` - -**Updated `create_initial_state()` method:** -- Changed `polling_id` → `id` -- Changed `object: "response.polling"` → `object: "response"` -- Replaced `content` (string) with `output` (array) -- Added `usage` field (null initially) -- Added `status_details` field -- Moved internal tracking to `_polling_state` object - -**Updated `update_state()` method:** -- Changed from updating `content` string to updating `output` array items -- Added support for `output_item` parameter -- Added support for `status_details` parameter -- Added support for `usage` parameter -- Structured error format with type/message/code - -**Updated `cancel_polling()` method:** -- Now sets status to `"cancelled"` with proper `status_details` - -#### B. `litellm/proxy/response_api_endpoints/endpoints.py` - -**Updated `_background_streaming_task()` function:** -- Processes OpenAI streaming events: - - `response.output_item.added` - - `response.content_part.added` - - `response.content_part.done` - - `response.output_item.done` - - `response.done` -- Builds output items incrementally -- Tracks output items by ID -- Extracts and stores usage data -- Sets proper status_details on completion - -**Updated `responses_api()` POST endpoint:** -- Returns OpenAI format response object instead of custom polling object -- Uses `response` as object type -- Sets `status: "in_progress"` initially -- Returns empty `output` array initially - -**Updated `responses_api()` GET endpoint:** -- Returns full OpenAI Response object structure -- Includes `output` array with items -- Includes `usage` if available -- Includes `status_details` - -### 4. Streaming Events Processing - -The background task now handles these OpenAI streaming events: - -1. **response.output_item.added**: Tracks new output items (messages, function calls) -2. **response.content_part.added**: Accumulates content parts as they stream -3. **response.content_part.done**: Finalizes content for an output item -4. **response.output_item.done**: Marks output item as complete -5. **response.done**: Finalizes response with usage data - -### 5. Redis Cache Structure - -**Cache Key:** `litellm:polling:response:litellm_poll_{uuid}` - -**Stored Object:** -```json -{ - "id": "litellm_poll_abc123", - "object": "response", - "status": "in_progress", - "status_details": null, - "output": [...], - "usage": null, - "metadata": {}, - "created_at": 1700000000, - "_polling_state": { - "updated_at": "2024-11-19T10:00:00Z", - "request_data": {...}, - "user_id": "user_123", - "team_id": "team_456", - "model": "gpt-4o", - "input": "..." - } -} -``` - -### 6. API Response Examples - -#### Starting Background Response - -**Request:** -```bash -curl -X POST http://localhost:4000/v1/responses \ - -H "Authorization: Bearer sk-1234" \ - -H "Content-Type: application/json" \ - -d '{ - "model": "gpt-4o", - "input": "Write an essay", - "background": true, - "metadata": {"user": "john"} - }' -``` - -**Response:** -```json -{ - "id": "litellm_poll_abc123", - "object": "response", - "status": "in_progress", - "status_details": null, - "output": [], - "usage": null, - "metadata": {"user": "john"}, - "created_at": 1700000000 -} -``` - -#### Polling for Updates - -**Request:** -```bash -curl -X GET http://localhost:4000/v1/responses/litellm_poll_abc123 \ - -H "Authorization: Bearer sk-1234" -``` - -**Response (In Progress):** -```json -{ - "id": "litellm_poll_abc123", - "object": "response", - "status": "in_progress", - "status_details": null, - "output": [ - { - "id": "item_001", - "type": "message", - "role": "assistant", - "status": "in_progress", - "content": [ - { - "type": "text", - "text": "Artificial intelligence is..." - } - ] - } - ], - "usage": null, - "metadata": {"user": "john"}, - "created_at": 1700000000 -} -``` - -**Response (Completed):** -```json -{ - "id": "litellm_poll_abc123", - "object": "response", - "status": "completed", - "status_details": { - "type": "completed", - "reason": "stop" - }, - "output": [ - { - "id": "item_001", - "type": "message", - "role": "assistant", - "status": "completed", - "content": [ - { - "type": "text", - "text": "Artificial intelligence is... [full essay]" - } - ] - } - ], - "usage": { - "input_tokens": 25, - "output_tokens": 1200, - "total_tokens": 1225 - }, - "metadata": {"user": "john"}, - "created_at": 1700000000 -} -``` - -### 7. Backward Compatibility Notes - -**Breaking Changes:** -- Field names changed (`polling_id` → `id`, `content` → `output`) -- Status values changed (`pending` → `in_progress`, `error` → `failed`) -- Error structure changed (nested under `status_details.error`) -- Content is now structured in `output` array instead of flat string - -**Migration Path:** -Clients need to: -1. Use `id` instead of `polling_id` -2. Parse `output` array to extract text content -3. Handle new status values -4. Read errors from `status_details.error` instead of top-level `error` - -### 8. Benefits of OpenAI Format - -1. **Standard Compliance**: Fully compatible with OpenAI's Response API -2. **Structured Output**: Supports multiple output types (messages, function calls) -3. **Better Streaming**: Aligned with OpenAI's streaming event format -4. **Token Tracking**: Built-in usage tracking -5. **Rich Status**: Detailed status information with reasons and error types -6. **Metadata Support**: Custom metadata at the response level - -### 9. Testing - -Updated `test_polling_feature.py` to: -- Validate OpenAI Response object structure -- Extract text from structured `output` array -- Check for proper status values -- Verify `usage` data -- Test `status_details` structure - -### 10. Documentation - -Created comprehensive documentation: -- **OPENAI_RESPONSE_FORMAT.md**: Complete format specification with examples -- **OPENAI_FORMAT_CHANGES_SUMMARY.md**: This file - summary of changes - -## Files Modified - -1. `litellm/proxy/response_polling/polling_handler.py` - Core polling handler -2. `litellm/proxy/response_api_endpoints/endpoints.py` - API endpoints -3. `test_polling_feature.py` - Test script -4. `litellm_config.yaml` - Configuration (no changes to format) - -## Files Created - -1. `OPENAI_RESPONSE_FORMAT.md` - Complete format documentation -2. `OPENAI_FORMAT_CHANGES_SUMMARY.md` - This summary document - -## Next Steps - -1. **Test with Real Providers**: Test streaming events with various LLM providers -2. **Client Libraries**: Update any client libraries to use new format -3. **Migration Guide**: Create guide for existing users -4. **Function Calling**: Test with function calling responses -5. **Performance**: Monitor Redis cache performance with structured objects - -## Validation Checklist - -- ✅ Response object follows OpenAI format -- ✅ Streaming events processed correctly -- ✅ Status values aligned with OpenAI -- ✅ Error format matches OpenAI structure -- ✅ Output items support multiple types -- ✅ Usage data captured and stored -- ✅ Metadata preserved throughout lifecycle -- ✅ Test script validates new format -- ✅ Documentation comprehensive and accurate -- ✅ Redis cache stores complete Response object - -## References - -- OpenAI Response API: https://platform.openai.com/docs/api-reference/responses -- OpenAI Streaming: https://platform.openai.com/docs/api-reference/responses-streaming -- LiteLLM Docs: https://docs.litellm.ai/ - diff --git a/OPENAI_RESPONSE_FORMAT.md b/OPENAI_RESPONSE_FORMAT.md deleted file mode 100644 index c00117798f..0000000000 --- a/OPENAI_RESPONSE_FORMAT.md +++ /dev/null @@ -1,523 +0,0 @@ -# OpenAI Response Object Format - Polling Via Cache Implementation - -## Overview - -The polling via cache feature now follows the official OpenAI Response object format as documented at: -- **Response Object**: https://platform.openai.com/docs/api-reference/responses/object -- **Streaming Events**: https://platform.openai.com/docs/api-reference/responses-streaming - -## Response Object Structure - -The Response object stored in Redis cache follows this structure: - -```json -{ - "id": "litellm_poll_abc123-def456", - "object": "response", - "status": "in_progress" | "completed" | "cancelled" | "failed" | "incomplete", - "status_details": { - "type": "completed" | "incomplete" | "cancelled" | "failed", - "reason": "stop" | "length" | "content_filter" | "user_requested", - "error": { - "type": "internal_error", - "message": "Error message", - "code": "error_code" - } - }, - "output": [ - { - "id": "item_001", - "type": "message", - "status": "completed", - "role": "assistant", - "content": [ - { - "type": "text", - "text": "Response content here..." - } - ] - } - ], - "usage": { - "input_tokens": 100, - "output_tokens": 500, - "total_tokens": 600 - }, - "metadata": { - "custom_field": "custom_value" - }, - "created_at": 1700000000 -} -``` - -### Internal Polling Fields - -For internal tracking, additional fields are stored under `_polling_state`: - -```json -{ - "_polling_state": { - "updated_at": "2024-11-19T10:00:05Z", - "request_data": { /* original request */ }, - "user_id": "user_123", - "team_id": "team_456", - "model": "gpt-4o", - "input": "User prompt..." - } -} -``` - -## Status Values - -Following OpenAI's format: - -| Status | Description | -|--------|-------------| -| `in_progress` | Response is currently being generated | -| `completed` | Response has been fully generated | -| `cancelled` | Response was cancelled by user | -| `failed` | Response generation failed with an error | -| `incomplete` | Response was cut off (length limit, content filter) | - -## Streaming Events Processing - -The background streaming task processes these OpenAI streaming events: - -### 1. `response.created` -Initial response created event (handled by initial state creation). - -### 2. `response.output_item.added` -```json -{ - "type": "response.output_item.added", - "item": { - "id": "item_001", - "type": "message", - "role": "assistant", - "status": "in_progress" - } -} -``` - -### 3. `response.content_part.added` -```json -{ - "type": "response.content_part.added", - "item_id": "item_001", - "output_index": 0, - "part": { - "type": "text", - "text": "Initial text..." - } -} -``` - -### 4. `response.content_part.done` -```json -{ - "type": "response.content_part.done", - "item_id": "item_001", - "part": { - "type": "text", - "text": "Complete text content" - } -} -``` - -### 5. `response.output_item.done` -```json -{ - "type": "response.output_item.done", - "item": { - "id": "item_001", - "type": "message", - "role": "assistant", - "status": "completed", - "content": [ - { - "type": "text", - "text": "Complete content" - } - ] - } -} -``` - -### 6. `response.done` -```json -{ - "type": "response.done", - "response": { - "id": "litellm_poll_abc123", - "status": "completed", - "status_details": { - "type": "completed", - "reason": "stop" - }, - "usage": { - "input_tokens": 100, - "output_tokens": 500, - "total_tokens": 600 - } - } -} -``` - -## API Examples - -### Creating a Background Response - -```bash -curl -X POST http://localhost:4000/v1/responses \ - -H "Authorization: Bearer sk-1234" \ - -H "Content-Type: application/json" \ - -d '{ - "model": "gpt-4o", - "input": "Write an essay about AI", - "background": true, - "metadata": { - "user": "john_doe", - "session_id": "sess_123" - } - }' -``` - -**Response:** -```json -{ - "id": "litellm_poll_abc123def456", - "object": "response", - "status": "in_progress", - "status_details": null, - "output": [], - "usage": null, - "metadata": { - "user": "john_doe", - "session_id": "sess_123" - }, - "created_at": 1700000000 -} -``` - -### Polling for Response (In Progress) - -```bash -curl -X GET http://localhost:4000/v1/responses/litellm_poll_abc123def456 \ - -H "Authorization: Bearer sk-1234" -``` - -**Response:** -```json -{ - "id": "litellm_poll_abc123def456", - "object": "response", - "status": "in_progress", - "status_details": null, - "output": [ - { - "id": "item_001", - "type": "message", - "role": "assistant", - "status": "in_progress", - "content": [ - { - "type": "text", - "text": "Artificial intelligence (AI) is a rapidly..." - } - ] - } - ], - "usage": null, - "metadata": { - "user": "john_doe", - "session_id": "sess_123" - }, - "created_at": 1700000000 -} -``` - -### Polling for Response (Completed) - -```bash -curl -X GET http://localhost:4000/v1/responses/litellm_poll_abc123def456 \ - -H "Authorization: Bearer sk-1234" -``` - -**Response:** -```json -{ - "id": "litellm_poll_abc123def456", - "object": "response", - "status": "completed", - "status_details": { - "type": "completed", - "reason": "stop" - }, - "output": [ - { - "id": "item_001", - "type": "message", - "role": "assistant", - "status": "completed", - "content": [ - { - "type": "text", - "text": "Artificial intelligence (AI) is a rapidly evolving field... [full essay]" - } - ] - } - ], - "usage": { - "input_tokens": 25, - "output_tokens": 1200, - "total_tokens": 1225 - }, - "metadata": { - "user": "john_doe", - "session_id": "sess_123" - }, - "created_at": 1700000000 -} -``` - -### Error Response - -```json -{ - "id": "litellm_poll_abc123def456", - "object": "response", - "status": "failed", - "status_details": { - "type": "failed", - "error": { - "type": "internal_error", - "message": "Provider timeout", - "code": "background_streaming_error" - } - }, - "output": [], - "usage": null, - "metadata": {}, - "created_at": 1700000000 -} -``` - -## Output Item Types - -### Message Output -```json -{ - "id": "item_001", - "type": "message", - "role": "assistant", - "status": "completed", - "content": [ - { - "type": "text", - "text": "Message content" - } - ] -} -``` - -### Function Call Output -```json -{ - "id": "item_002", - "type": "function_call", - "status": "completed", - "name": "get_weather", - "call_id": "call_abc123", - "arguments": "{\"location\": \"San Francisco\"}" -} -``` - -### Function Call Output Result -```json -{ - "id": "item_003", - "type": "function_call_output", - "call_id": "call_abc123", - "output": "{\"temperature\": 72, \"condition\": \"sunny\"}" -} -``` - -## Redis Cache Storage - -### Key Format -``` -litellm:polling:response:litellm_poll_{uuid} -``` - -### TTL -- Default: 3600 seconds (1 hour) -- Configurable via `ttl` parameter - -### Storage Example -```redis -> KEYS litellm:polling:response:* -1) "litellm:polling:response:litellm_poll_abc123def456" - -> GET "litellm:polling:response:litellm_poll_abc123def456" -"{\"id\":\"litellm_poll_abc123def456\",\"object\":\"response\",\"status\":\"completed\",...}" - -> TTL "litellm:polling:response:litellm_poll_abc123def456" -(integer) 2847 -``` - -## Client Implementation Example - -### Python Client - -```python -import time -import requests - -def poll_response(polling_id, api_key): - """Poll for response following OpenAI format""" - url = f"http://localhost:4000/v1/responses/{polling_id}" - headers = {"Authorization": f"Bearer {api_key}"} - - while True: - response = requests.get(url, headers=headers) - data = response.json() - - status = data["status"] - print(f"Status: {status}") - - # Extract content from output items - for item in data.get("output", []): - if item["type"] == "message": - content = "" - for part in item.get("content", []): - if part["type"] == "text": - content += part["text"] - print(f"Content: {content[:100]}...") - - # Check status - if status == "completed": - print("\n✅ Response completed!") - print(f"Usage: {data.get('usage')}") - return data - elif status == "failed": - error = data.get("status_details", {}).get("error", {}) - print(f"\n❌ Error: {error.get('message')}") - return None - elif status == "cancelled": - print("\n⚠️ Response cancelled") - return None - - time.sleep(2) # Poll every 2 seconds - -# Start background response -response = requests.post( - "http://localhost:4000/v1/responses", - headers={ - "Authorization": "Bearer sk-1234", - "Content-Type": "application/json" - }, - json={ - "model": "gpt-4o", - "input": "Write an essay", - "background": True - } -) - -polling_id = response.json()["id"] -result = poll_response(polling_id, "sk-1234") -``` - -### JavaScript/TypeScript Client - -```typescript -interface ResponseObject { - id: string; - object: "response"; - status: "in_progress" | "completed" | "cancelled" | "failed" | "incomplete"; - status_details: { - type: string; - reason?: string; - error?: { - type: string; - message: string; - code: string; - }; - } | null; - output: Array<{ - id: string; - type: "message" | "function_call" | "function_call_output"; - content?: Array<{ type: "text"; text: string }>; - [key: string]: any; - }>; - usage: { - input_tokens: number; - output_tokens: number; - total_tokens: number; - } | null; - metadata: Record; - created_at: number; -} - -async function pollResponse(pollingId: string, apiKey: string): Promise { - const url = `http://localhost:4000/v1/responses/${pollingId}`; - const headers = { Authorization: `Bearer ${apiKey}` }; - - while (true) { - const response = await fetch(url, { headers }); - const data: ResponseObject = await response.json(); - - console.log(`Status: ${data.status}`); - - // Extract text content - for (const item of data.output) { - if (item.type === "message" && item.content) { - const text = item.content - .filter(p => p.type === "text") - .map(p => p.text) - .join(""); - console.log(`Content: ${text.substring(0, 100)}...`); - } - } - - if (data.status === "completed") { - console.log("✅ Response completed!"); - console.log("Usage:", data.usage); - return data; - } else if (data.status === "failed") { - throw new Error(data.status_details?.error?.message || "Unknown error"); - } else if (data.status === "cancelled") { - throw new Error("Response was cancelled"); - } - - await new Promise(resolve => setTimeout(resolve, 2000)); - } -} -``` - -## Compatibility Notes - -1. **OpenAI API Compatibility**: The response format is fully compatible with OpenAI's Response API -2. **Polling ID Prefix**: The `litellm_poll_` prefix allows the proxy to distinguish between polling IDs and provider response IDs -3. **Internal Fields**: The `_polling_state` object is for internal use only and not exposed in the API response -4. **Provider Agnostic**: Works with any LLM provider through LiteLLM's unified interface - -## Migration from Previous Format - -If you were using the previous format, here are the key changes: - -| Old Field | New Field | Notes | -|-----------|-----------|-------| -| `polling_id` | `id` | Standard field name | -| `object: "response.polling"` | `object: "response"` | OpenAI format | -| `status: "pending"` | `status: "in_progress"` | Aligned with OpenAI | -| `status: "streaming"` | `status: "in_progress"` | Same as above | -| `content` | `output[].content[]` | Structured output items | -| `error` | `status_details.error` | Nested error object | -| N/A | `usage` | Added token usage tracking | - -## References - -- OpenAI Response Object: https://platform.openai.com/docs/api-reference/responses/object -- OpenAI Response Streaming: https://platform.openai.com/docs/api-reference/responses-streaming -- LiteLLM Documentation: https://docs.litellm.ai/ - diff --git a/POLLING_VIA_CACHE_FEATURE.md b/POLLING_VIA_CACHE_FEATURE.md deleted file mode 100644 index 88c58f4baa..0000000000 --- a/POLLING_VIA_CACHE_FEATURE.md +++ /dev/null @@ -1,413 +0,0 @@ -# Polling Via Cache Feature - -## Overview - -The Polling Via Cache feature allows users to make background Response API calls that return immediately with a polling ID, while the actual LLM response is streamed in the background and cached in Redis. Clients can poll the cached response to retrieve partial or complete results. - -## Configuration - -Add the following to your `litellm_config.yaml`: - -```yaml -litellm_settings: - cache: true - cache_params: - type: redis - ttl: 3600 - host: "127.0.0.1" - port: "6379" - - # Response API polling configuration - responses: - background_mode: - # Enable polling via cache for background responses - # Options: - # - "all" or ["all"]: Enable for all models - # - ["gpt-4o", "gpt-4"]: Enable for specific models - # - ["openai", "anthropic"]: Enable for specific providers - polling_via_cache: ["all"] -``` - -## How It Works - -### 1. Request Flow - -When `background=true` is set in a Response API request: - -1. **Detection**: Proxy checks if polling_via_cache is enabled and Redis is available -2. **UUID Generation**: Creates a polling ID with prefix `litellm_poll_` -3. **Initial State**: Stores initial state in Redis (TTL: 1 hour) -4. **Background Task**: Starts async task to stream response and update cache -5. **Immediate Return**: Returns polling ID to client - -### 2. Background Streaming - -The background task: -- Forces `stream=true` on the request -- Streams the response from the provider -- Updates Redis cache with cumulative content -- Stores final response when complete -- Handles errors and stores them in cache - -### 3. Polling - -Clients use the existing GET endpoint with the polling ID: -- Proxy detects `litellm_poll_` prefix -- Returns cached state instead of calling provider -- Includes cumulative content, status, and metadata - -## API Usage - -### 1. Start Background Response - -```bash -curl -X POST http://localhost:4000/v1/responses \ - -H "Authorization: Bearer sk-1234" \ - -H "Content-Type: application/json" \ - -d '{ - "model": "gpt-4o", - "input": "Write a long essay about artificial intelligence", - "background": true - }' -``` - -**Response:** -```json -{ - "id": "litellm_poll_abc123def456", - "object": "response.polling", - "status": "pending", - "created_at": 1700000000, - "message": "Response is being generated in background. Use GET /v1/responses/{id} to retrieve partial or complete response." -} -``` - -### 2. Poll for Response - -```bash -curl -X GET http://localhost:4000/v1/responses/litellm_poll_abc123def456 \ - -H "Authorization: Bearer sk-1234" -``` - -**Response (while streaming):** -```json -{ - "id": "litellm_poll_abc123def456", - "object": "response.polling", - "status": "streaming", - "created_at": "2024-11-19T10:00:00Z", - "updated_at": "2024-11-19T10:00:05Z", - "content": "Artificial intelligence (AI) is a rapidly evolving field...", - "content_length": 500, - "chunk_count": 15, - "metadata": { - "model": "gpt-4o", - "input": "Write a long essay about artificial intelligence" - }, - "error": null, - "final_response": null -} -``` - -**Response (completed):** -```json -{ - "id": "litellm_poll_abc123def456", - "object": "response.polling", - "status": "completed", - "created_at": "2024-11-19T10:00:00Z", - "updated_at": "2024-11-19T10:00:30Z", - "content": "Artificial intelligence (AI) is a rapidly evolving field... [full essay]", - "content_length": 5000, - "chunk_count": 150, - "metadata": { - "model": "gpt-4o", - "input": "Write a long essay about artificial intelligence" - }, - "error": null, - "final_response": { /* OpenAI response object */ } -} -``` - -### 3. Delete/Cancel Response - -```bash -curl -X DELETE http://localhost:4000/v1/responses/litellm_poll_abc123def456 \ - -H "Authorization: Bearer sk-1234" -``` - -**Response:** -```json -{ - "id": "litellm_poll_abc123def456", - "object": "response.deleted", - "deleted": true -} -``` - -## Status Values - -| Status | Description | -|--------|-------------| -| `pending` | Request received, background task not yet started | -| `streaming` | Background task is actively streaming response | -| `completed` | Response fully generated and cached | -| `error` | An error occurred during generation | -| `cancelled` | Response was cancelled by user | - -## Implementation Details - -### Polling ID Format - -- **Prefix**: `litellm_poll_` -- **Format**: `litellm_poll_{uuid}` -- **Example**: `litellm_poll_abc123-def456-789ghi` - -This prefix allows the GET endpoint to distinguish between: -- Polling IDs (handled by Redis cache) -- Provider response IDs (passed through to provider API) - -### Redis Cache Structure - -**Key**: `litellm:polling:response:litellm_poll_{uuid}` - -**Value** (JSON): -```json -{ - "polling_id": "litellm_poll_abc123", - "object": "response.polling", - "status": "streaming", - "created_at": "2024-11-19T10:00:00Z", - "updated_at": "2024-11-19T10:00:05Z", - "request_data": { /* original request */ }, - "user_id": "user_123", - "team_id": "team_456", - "content": "cumulative content so far...", - "chunks": [ /* all streaming chunks */ ], - "metadata": { - "model": "gpt-4o", - "input": "..." - }, - "error": null, - "final_response": null -} -``` - -**TTL**: 3600 seconds (1 hour) - -### Security - -- User/Team ID verification on GET and DELETE -- Only the user who created the request (or team members) can access it -- Automatic expiry after 1 hour prevents stale data - -## Configuration Options - -### Enable for All Models - -```yaml -responses: - background_mode: - polling_via_cache: ["all"] -``` - -### Enable for Specific Models - -```yaml -responses: - background_mode: - polling_via_cache: ["gpt-4o", "gpt-4", "claude-3"] -``` - -### Enable for Specific Providers - -```yaml -responses: - background_mode: - polling_via_cache: ["openai", "anthropic"] -``` - -This will match any model starting with `openai/` or `anthropic/`. - -## Benefits - -1. **Immediate Response**: Client gets polling ID instantly, no waiting -2. **Partial Results**: Can retrieve partial content while generation continues -3. **Progress Monitoring**: Poll at intervals to show progress to users -4. **Error Handling**: Errors are cached and can be retrieved -5. **Scalability**: Background tasks don't block API requests - -## Limitations - -1. **Requires Redis**: Feature only works with Redis cache configured -2. **1 Hour TTL**: Responses expire after 1 hour -3. **No Streaming to Client**: Client must poll, no real-time streaming -4. **Memory Usage**: Full response stored in Redis - -## Example Client Implementation - -### Python - -```python -import time -import requests - -# Start background response -response = requests.post( - "http://localhost:4000/v1/responses", - headers={"Authorization": "Bearer sk-1234"}, - json={ - "model": "gpt-4o", - "input": "Write a long essay", - "background": True - } -) - -polling_id = response.json()["id"] -print(f"Started background response: {polling_id}") - -# Poll for results -while True: - poll_response = requests.get( - f"http://localhost:4000/v1/responses/{polling_id}", - headers={"Authorization": "Bearer sk-1234"} - ) - - data = poll_response.json() - status = data["status"] - content = data["content"] - - print(f"Status: {status}, Content length: {len(content)}") - - if status == "completed": - print("Final response:", content) - break - elif status == "error": - print("Error:", data["error"]) - break - - time.sleep(2) # Poll every 2 seconds -``` - -### JavaScript - -```javascript -async function pollResponse(pollingId) { - while (true) { - const response = await fetch( - `http://localhost:4000/v1/responses/${pollingId}`, - { headers: { 'Authorization': 'Bearer sk-1234' } } - ); - - const data = await response.json(); - console.log(`Status: ${data.status}, Content: ${data.content.substring(0, 50)}...`); - - if (data.status === 'completed') { - console.log('Final response:', data.content); - break; - } else if (data.status === 'error') { - console.error('Error:', data.error); - break; - } - - await new Promise(resolve => setTimeout(resolve, 2000)); // Wait 2s - } -} - -// Start background response -const startResponse = await fetch('http://localhost:4000/v1/responses', { - method: 'POST', - headers: { - 'Authorization': 'Bearer sk-1234', - 'Content-Type': 'application/json' - }, - body: JSON.stringify({ - model: 'gpt-4o', - input: 'Write a long essay', - background: true - }) -}); - -const { id } = await startResponse.json(); -await pollResponse(id); -``` - -## Testing - -To test the feature: - -1. **Start Redis** (if not already running): - ```bash - redis-server --port 6379 - ``` - -2. **Start LiteLLM Proxy**: - ```bash - python -m litellm.proxy.proxy_cli --config litellm_config.yaml --detailed_debug - ``` - -3. **Make a background request**: - ```bash - curl -X POST http://localhost:4000/v1/responses \ - -H "Authorization: Bearer sk-test-key" \ - -H "Content-Type: application/json" \ - -d '{ - "model": "gpt-4o", - "input": "Count from 1 to 100", - "background": true - }' - ``` - -4. **Poll for results**: - ```bash - # Replace with your polling_id - curl http://localhost:4000/v1/responses/litellm_poll_XXX \ - -H "Authorization: Bearer sk-test-key" - ``` - -5. **Check Redis**: - ```bash - redis-cli - > KEYS litellm:polling:response:* - > GET litellm:polling:response:litellm_poll_XXX - ``` - -## Troubleshooting - -### Issue: Polling not enabled - -**Symptom**: Requests with `background=true` return immediately without streaming - -**Solution**: -- Verify Redis is running and accessible -- Check `redis_usage_cache` is initialized -- Ensure `polling_via_cache` is configured - -### Issue: Polling ID not found - -**Symptom**: GET returns 404 - -**Possible causes**: -- Response expired (>1 hour old) -- Redis connection lost -- Wrong polling ID - -### Issue: Empty content - -**Symptom**: Content length is 0 - -**Possible causes**: -- Background task still starting -- Error in streaming -- Check logs for background task errors - -## Future Enhancements - -Potential improvements: -1. WebSocket support for real-time updates -2. Configurable TTL per request -3. Compression for large responses -4. Pagination for very long responses -5. Metrics and monitoring endpoints - - diff --git a/REFACTOR_NATIVE_OPENAI_TYPES.md b/REFACTOR_NATIVE_OPENAI_TYPES.md deleted file mode 100644 index 5a167f986c..0000000000 --- a/REFACTOR_NATIVE_OPENAI_TYPES.md +++ /dev/null @@ -1,309 +0,0 @@ -# Refactoring to Native OpenAI Types - -## Summary - -Successfully refactored the polling via cache implementation to use OpenAI's native types from `litellm.types.llms.openai` instead of custom implementations. - -## Changes Made - -### 1. Removed Custom `ResponseState` Class ❌ - -**Before:** -```python -class ResponseState: - """Enum-like class for polling states""" - QUEUED = "queued" - IN_PROGRESS = "in_progress" - COMPLETED = "completed" - CANCELLED = "cancelled" - FAILED = "failed" - INCOMPLETE = "incomplete" -``` - -**After:** ✅ Using OpenAI's native `ResponsesAPIStatus` type -```python -from litellm.types.llms.openai import ResponsesAPIResponse, ResponsesAPIStatus - -# ResponsesAPIStatus is defined as: -# Literal["completed", "failed", "in_progress", "cancelled", "queued", "incomplete"] -``` - -### 2. Using `ResponsesAPIResponse` Object - -**Before - Manual Dict Construction:** -```python -initial_state = { - "id": polling_id, - "object": "response", - "status": ResponseState.QUEUED, - "status_details": None, - "output": [], - "usage": None, - "metadata": request_data.get("metadata", {}), - "created_at": created_timestamp, - "_polling_state": {...} -} -``` - -**After - Using OpenAI Type:** -```python -# Create OpenAI-compliant response object -response = ResponsesAPIResponse( - id=polling_id, - object="response", - status="queued", # Native OpenAI status value - created_at=created_timestamp, - output=[], - metadata=request_data.get("metadata", {}), - usage=None, -) - -# Serialize to dict and add internal state for cache -cache_data = { - **response.dict(), # Pydantic serialization - "_polling_state": {...} -} -``` - -### 3. Updated Method Signatures - -**`create_initial_state()` Return Type:** -```python -# Before -async def create_initial_state(...) -> Dict[str, Any]: - -# After -async def create_initial_state(...) -> ResponsesAPIResponse: -``` - -**`update_state()` Parameter Type:** -```python -# Before -async def update_state( - self, - polling_id: str, - status: Optional[str] = None, - ... -) - -# After -async def update_state( - self, - polling_id: str, - status: Optional[ResponsesAPIStatus] = None, # Type-safe! - ... -) -``` - -### 4. Status Values Now Type-Safe - -All status values are now validated by TypeScript/Pydantic: - -```python -# Valid status values (enforced by ResponsesAPIStatus type) -"queued" # ✅ -"in_progress" # ✅ -"completed" # ✅ -"cancelled" # ✅ -"failed" # ✅ -"incomplete" # ✅ - -# Invalid values will be caught by type checker -"pending" # ❌ Type error! -"error" # ❌ Type error! -``` - -## Benefits - -### ✅ Type Safety -- Pydantic validation ensures correct field types -- Status values are type-checked -- IDE auto-completion works perfectly - -### ✅ OpenAI Compatibility -- Guaranteed to match OpenAI's Response API spec -- Automatic updates when OpenAI types are updated -- No drift between our implementation and OpenAI's spec - -### ✅ Better Developer Experience -- Full IDE support with auto-completion -- Type hints for all fields -- Self-documenting code - -### ✅ Built-in Serialization -- `.dict()` method for JSON serialization -- `.json()` method for direct JSON string -- Proper handling of Optional fields - -### ✅ Validation -- Automatic field validation via Pydantic -- Type coercion where appropriate -- Clear error messages on invalid data - -## File Changes - -### Modified Files: - -1. **`litellm/proxy/response_polling/polling_handler.py`** - - ✅ Removed custom `ResponseState` class - - ✅ Added imports: `ResponsesAPIResponse`, `ResponsesAPIStatus` - - ✅ Updated `create_initial_state()` to return `ResponsesAPIResponse` - - ✅ Updated `update_state()` to use `ResponsesAPIStatus` type - - ✅ All status strings are now native OpenAI values - -2. **`litellm/proxy/response_api_endpoints/endpoints.py`** - - ✅ Removed `ResponseState` import - - ✅ Status strings used directly ("queued", "in_progress", etc.) - -### No Breaking Changes for API Consumers - -The API response format remains identical: -```json -{ - "id": "litellm_poll_abc123", - "object": "response", - "status": "queued", - "output": [], - "usage": null, - "metadata": {}, - "created_at": 1700000000 -} -``` - -## Type Definitions Used - -### From `litellm/types/llms/openai.py`: - -```python -# Status type -ResponsesAPIStatus = Literal[ - "completed", "failed", "in_progress", "cancelled", "queued", "incomplete" -] - -# Response object -class ResponsesAPIResponse(BaseLiteLLMOpenAIResponseObject): - id: str - created_at: int - error: Optional[dict] = None - incomplete_details: Optional[IncompleteDetails] = None - instructions: Optional[str] = None - metadata: Optional[Dict] = None - model: Optional[str] = None - object: Optional[str] = None - output: Union[List[Union[ResponseOutputItem, Dict]], ...] - status: Optional[str] = None - usage: Optional[ResponseAPIUsage] = None - # ... and more fields -``` - -## Usage Example - -### Creating a Response: - -```python -from litellm.types.llms.openai import ResponsesAPIResponse - -# Type-safe creation -response = ResponsesAPIResponse( - id="litellm_poll_abc123", - object="response", - status="queued", # Auto-validated! - created_at=1700000000, - output=[], - metadata={"user": "test"}, - usage=None, -) - -# Serialize to dict -response_dict = response.dict() - -# Serialize to JSON string -response_json = response.json() -``` - -### Updating Status: - -```python -# Type-safe status updates -await polling_handler.update_state( - polling_id="litellm_poll_abc123", - status="in_progress", # IDE will suggest valid values! -) - -# Invalid status would be caught by type checker -await polling_handler.update_state( - polling_id="litellm_poll_abc123", - status="streaming", # ❌ Type error - not a valid ResponsesAPIStatus -) -``` - -## Migration Notes - -### For Developers: - -1. **No more custom status constants**: Use string literals directly - ```python - # Old - status = ResponseState.QUEUED - - # New - status = "queued" # Type-safe with ResponsesAPIStatus - ``` - -2. **Type hints work**: Your IDE will now suggest valid status values - -3. **Validation is automatic**: Invalid values are caught at runtime by Pydantic - -### For API Consumers: - -No changes required! The API response format is identical. - -## Testing - -All existing tests continue to work without modification: - -```python -# Test still works -response = await client.post("/v1/responses", json={ - "model": "gpt-4o", - "input": "test", - "background": True -}) - -assert response["status"] == "queued" # ✅ Still valid -assert response["object"] == "response" # ✅ Still valid -``` - -## Future Improvements - -1. **Consider using Pydantic models throughout**: Extend this pattern to other parts of the codebase - -2. **Add status transition validation**: Ensure only valid status transitions (e.g., queued → in_progress → completed) - -3. **Use TypedDict for internal state**: Type-safe `_polling_state` object - -4. **Add response builders**: Helper methods for common response patterns - -## Validation Checklist - -- ✅ All status values use OpenAI native types -- ✅ Response objects use `ResponsesAPIResponse` -- ✅ Type hints are correct throughout -- ✅ No linting errors -- ✅ No breaking changes to API -- ✅ Backward compatible with existing code -- ✅ IDE auto-completion works -- ✅ Documentation updated - -## References - -- OpenAI Response API: https://platform.openai.com/docs/api-reference/responses/object -- LiteLLM OpenAI Types: `litellm/types/llms/openai.py` -- Pydantic Documentation: https://docs.pydantic.dev/ - ---- - -**Status**: ✅ Complete -**Date**: 2024-11-19 -**Impact**: Internal refactoring, no API changes - diff --git a/litellm/proxy/response_api_endpoints/endpoints.py b/litellm/proxy/response_api_endpoints/endpoints.py index b5b10c440f..6517b5ddc7 100644 --- a/litellm/proxy/response_api_endpoints/endpoints.py +++ b/litellm/proxy/response_api_endpoints/endpoints.py @@ -1,7 +1,9 @@ -from fastapi import APIRouter, Depends, HTTPException, Request, Response +import asyncio import json from typing import Any, Dict +from fastapi import APIRouter, Depends, HTTPException, Request, Response + from litellm._logging import verbose_proxy_logger from litellm.proxy._types import * from litellm.proxy.auth.user_api_key_auth import UserAPIKeyAuth, user_api_key_auth @@ -76,8 +78,31 @@ async def _background_streaming_task( ) # Process streaming response following OpenAI events format + # https://platform.openai.com/docs/api-reference/responses-streaming output_items = {} # Track output items by ID + accumulated_text = {} # Track accumulated text deltas by (output_index, content_index) usage_data = None + reasoning_data = None + tool_choice_data = None + tools_data = None + state_dirty = False # Track if state needs to be synced + last_update_time = asyncio.get_event_loop().time() + UPDATE_INTERVAL = 0.150 # 150ms batching interval + + async def flush_state_if_needed(force: bool = False) -> None: + """Flush accumulated state to Redis if interval elapsed or forced""" + nonlocal state_dirty, last_update_time + + current_time = asyncio.get_event_loop().time() + if state_dirty and (force or (current_time - last_update_time) >= UPDATE_INTERVAL): + # Convert output_items dict to list for update + output_list = list(output_items.values()) + await polling_handler.update_state( + polling_id=polling_id, + output=output_list, + ) + state_dirty = False + last_update_time = current_time # Handle StreamingResponse if hasattr(response, 'body_iterator'): @@ -95,22 +120,18 @@ async def _background_streaming_task( event = json.loads(chunk_data) event_type = event.get("type", "") - # Process different event types + # Process different event types based on OpenAI streaming spec if event_type == "response.output_item.added": # New output item added item = event.get("item", {}) item_id = item.get("id") if item_id: output_items[item_id] = item - await polling_handler.update_state( - polling_id=polling_id, - output_item=item, - ) + state_dirty = True elif event_type == "response.content_part.added": # Content part added to an output item item_id = event.get("item_id") - output_index = event.get("output_index") content_part = event.get("part", {}) if item_id and item_id in output_items: @@ -118,69 +139,100 @@ async def _background_streaming_task( if "content" not in output_items[item_id]: output_items[item_id]["content"] = [] output_items[item_id]["content"].append(content_part) + state_dirty = True + + elif event_type == "response.output_text.delta": + # Text delta - accumulate text content + # https://platform.openai.com/docs/api-reference/responses-streaming/response-text-delta + item_id = event.get("item_id") + output_index = event.get("output_index", 0) + content_index = event.get("content_index", 0) + delta = event.get("delta", "") + + if item_id and item_id in output_items: + # Accumulate text delta + key = (item_id, content_index) + if key not in accumulated_text: + accumulated_text[key] = "" + accumulated_text[key] += delta - await polling_handler.update_state( - polling_id=polling_id, - output_item=output_items[item_id], - ) + # Update the content in output_items + if "content" in output_items[item_id]: + content_list = output_items[item_id]["content"] + if content_index < len(content_list): + # Update existing content part with accumulated text + if isinstance(content_list[content_index], dict): + content_list[content_index]["text"] = accumulated_text[key] + state_dirty = True elif event_type == "response.content_part.done": # Content part completed item_id = event.get("item_id") content_part = event.get("part", {}) + content_index = event.get("content_index", 0) if item_id and item_id in output_items: - # Update final content - output_items[item_id]["content"] = content_part.get("content", "") - await polling_handler.update_state( - polling_id=polling_id, - output_item=output_items[item_id], - ) + # Update with final content from event + if "content" in output_items[item_id]: + content_list = output_items[item_id]["content"] + if content_index < len(content_list): + content_list[content_index] = content_part + state_dirty = True elif event_type == "response.output_item.done": - # Output item completed + # Output item completed - use final item data item = event.get("item", {}) item_id = item.get("id") if item_id: output_items[item_id] = item - await polling_handler.update_state( - polling_id=polling_id, - output_item=item, - ) + state_dirty = True - elif event_type == "response.done": - # Response completed - includes usage + elif event_type == "response.in_progress": + # Response is now in progress + # https://platform.openai.com/docs/api-reference/responses-streaming/response-in-progress + await polling_handler.update_state( + polling_id=polling_id, + status="in_progress", + ) + + elif event_type == "response.completed": + # Response completed - includes usage, reasoning, tools, tool_choice + # https://platform.openai.com/docs/api-reference/responses-streaming/response-completed response_data = event.get("response", {}) usage_data = response_data.get("usage") - - # Handle generic response format (for non-OpenAI providers) - elif "output" in event: - output = event.get("output", []) - if isinstance(output, list): - for item in output: + reasoning_data = response_data.get("reasoning") + tool_choice_data = response_data.get("tool_choice") + tools_data = response_data.get("tools") + + # Also update output from final response if available + if "output" in response_data: + final_output = response_data.get("output", []) + for item in final_output: item_id = item.get("id") if item_id: output_items[item_id] = item - await polling_handler.update_state( - polling_id=polling_id, - output_item=item, - ) - - # Check for usage in generic format - if "usage" in event: - usage_data = event.get("usage") + state_dirty = True + + # Flush state to Redis if interval elapsed + await flush_state_if_needed() except json.JSONDecodeError as e: verbose_proxy_logger.warning( f"Failed to parse streaming chunk: {e}" ) pass + + # Final flush to ensure all accumulated state is saved + await flush_state_if_needed(force=True) - # Mark as completed + # Mark as completed with all response data await polling_handler.update_state( polling_id=polling_id, status="completed", usage=usage_data, + reasoning=reasoning_data, + tool_choice=tool_choice_data, + tools=tools_data, ) verbose_proxy_logger.info( diff --git a/litellm/proxy/response_polling/polling_handler.py b/litellm/proxy/response_polling/polling_handler.py index 6475ee57cc..0412c2ff2e 100644 --- a/litellm/proxy/response_polling/polling_handler.py +++ b/litellm/proxy/response_polling/polling_handler.py @@ -87,10 +87,13 @@ class ResponsePollingHandler: self, polling_id: str, status: Optional[ResponsesAPIStatus] = None, - output_item: Optional[Dict] = None, usage: Optional[Dict] = None, error: Optional[Dict] = None, incomplete_details: Optional[Dict] = None, + reasoning: Optional[Dict] = None, + tool_choice: Optional[Any] = None, + tools: Optional[list] = None, + output: Optional[list] = None, ) -> None: """ Update the polling state in Redis @@ -101,10 +104,13 @@ class ResponsePollingHandler: Args: polling_id: Unique identifier for this polling request status: OpenAI ResponsesAPIStatus value - output_item: Output item to add/update usage: Usage information error: Error dict (automatically sets status to "failed") incomplete_details: Details for incomplete responses + reasoning: Reasoning configuration from response.completed + tool_choice: Tool choice configuration from response.completed + tools: Tools list from response.completed + output: Full output list to replace current output """ if not self.redis_cache: return @@ -126,22 +132,9 @@ class ResponsePollingHandler: if status: state["status"] = status - # Add output item (e.g., message, function_call) - if output_item: - # Check if we're updating an existing output item or adding new - item_id = output_item.get("id") - if item_id: - # Update existing item - found = False - for i, existing_item in enumerate(state["output"]): - if existing_item.get("id") == item_id: - state["output"][i] = output_item - found = True - break - if not found: - state["output"].append(output_item) - else: - state["output"].append(output_item) + # Replace full output list if provided + if output is not None: + state["output"] = output # Update usage if usage: @@ -156,6 +149,14 @@ class ResponsePollingHandler: if incomplete_details: state["incomplete_details"] = incomplete_details + # Update reasoning, tool_choice, tools from response.completed + if reasoning is not None: + state["reasoning"] = reasoning + if tool_choice is not None: + state["tool_choice"] = tool_choice + if tools is not None: + state["tools"] = tools + # Update cache with configured TTL await self.redis_cache.async_set_cache( key=cache_key, diff --git a/tests/proxy_unit_tests/test_response_polling_handler.py b/tests/proxy_unit_tests/test_response_polling_handler.py new file mode 100644 index 0000000000..352fe3e424 --- /dev/null +++ b/tests/proxy_unit_tests/test_response_polling_handler.py @@ -0,0 +1,530 @@ +""" +Unit tests for ResponsePollingHandler + +Tests core functionality including: +1. Polling ID generation and detection +2. Initial state creation (queued status) +3. State updates with batched output +4. Status transitions (queued -> in_progress -> completed) +5. Response completion with reasoning, tools, tool_choice +6. Error handling and cancellation +7. Cache key generation + +These tests ensure the polling handler correctly manages response state +following the OpenAI Response API format. +""" + +import json +import os +import sys +from datetime import datetime, timezone +from typing import Any, Dict, Optional +from unittest.mock import AsyncMock, Mock, patch + +import pytest + +sys.path.insert(0, os.path.abspath("../..")) + +from litellm.proxy.response_polling.polling_handler import ResponsePollingHandler + + +class TestResponsePollingHandler: + """Test cases for ResponsePollingHandler""" + + # ==================== Polling ID Tests ==================== + + def test_generate_polling_id_has_correct_prefix(self): + """Test that generated polling IDs have the correct prefix""" + polling_id = ResponsePollingHandler.generate_polling_id() + + assert polling_id.startswith("litellm_poll_") + assert len(polling_id) > len("litellm_poll_") # Has UUID after prefix + + def test_generate_polling_id_is_unique(self): + """Test that each generated polling ID is unique""" + ids = [ResponsePollingHandler.generate_polling_id() for _ in range(100)] + + assert len(ids) == len(set(ids)) # All unique + + def test_is_polling_id_returns_true_for_polling_ids(self): + """Test that is_polling_id correctly identifies polling IDs""" + polling_id = ResponsePollingHandler.generate_polling_id() + + assert ResponsePollingHandler.is_polling_id(polling_id) is True + + def test_is_polling_id_returns_false_for_provider_ids(self): + """Test that is_polling_id returns False for provider response IDs""" + # OpenAI format + assert ResponsePollingHandler.is_polling_id("resp_abc123") is False + # Anthropic format + assert ResponsePollingHandler.is_polling_id("msg_01XFDUDYJgAACzvnptvVoYEL") is False + # Generic UUID + assert ResponsePollingHandler.is_polling_id("550e8400-e29b-41d4-a716-446655440000") is False + + def test_get_cache_key_format(self): + """Test that cache keys have the correct format""" + polling_id = "litellm_poll_abc123" + cache_key = ResponsePollingHandler.get_cache_key(polling_id) + + assert cache_key == "litellm:polling:response:litellm_poll_abc123" + + # ==================== Initial State Tests ==================== + + @pytest.mark.asyncio + async def test_create_initial_state_returns_queued_status(self): + """Test that create_initial_state returns response with queued status""" + mock_redis = AsyncMock() + handler = ResponsePollingHandler(redis_cache=mock_redis, ttl=3600) + + polling_id = "litellm_poll_test123" + request_data = { + "model": "gpt-4o", + "input": "Hello", + "metadata": {"test": "value"} + } + + response = await handler.create_initial_state( + polling_id=polling_id, + request_data=request_data, + ) + + assert response.id == polling_id + assert response.object == "response" + assert response.status == "queued" + assert response.output == [] + assert response.usage is None + assert response.metadata == {"test": "value"} + + @pytest.mark.asyncio + async def test_create_initial_state_stores_in_redis(self): + """Test that create_initial_state stores state in Redis with correct TTL""" + mock_redis = AsyncMock() + handler = ResponsePollingHandler(redis_cache=mock_redis, ttl=7200) + + polling_id = "litellm_poll_test123" + request_data = {"model": "gpt-4o", "input": "Hello"} + + await handler.create_initial_state( + polling_id=polling_id, + request_data=request_data, + ) + + # Verify Redis was called with correct parameters + mock_redis.async_set_cache.assert_called_once() + call_args = mock_redis.async_set_cache.call_args + + assert call_args.kwargs["key"] == "litellm:polling:response:litellm_poll_test123" + assert call_args.kwargs["ttl"] == 7200 + + # Verify the stored value is valid JSON + stored_value = call_args.kwargs["value"] + parsed = json.loads(stored_value) + assert parsed["id"] == polling_id + assert parsed["status"] == "queued" + + @pytest.mark.asyncio + async def test_create_initial_state_sets_created_at_timestamp(self): + """Test that create_initial_state sets a valid created_at timestamp""" + mock_redis = AsyncMock() + handler = ResponsePollingHandler(redis_cache=mock_redis) + + before_time = int(datetime.now(timezone.utc).timestamp()) + + response = await handler.create_initial_state( + polling_id="litellm_poll_test", + request_data={}, + ) + + after_time = int(datetime.now(timezone.utc).timestamp()) + + assert before_time <= response.created_at <= after_time + + # ==================== State Update Tests ==================== + + @pytest.mark.asyncio + async def test_update_state_changes_status_to_in_progress(self): + """Test that update_state can change status to in_progress""" + mock_redis = AsyncMock() + mock_redis.async_get_cache.return_value = json.dumps({ + "id": "litellm_poll_test", + "object": "response", + "status": "queued", + "output": [], + "created_at": 1234567890 + }) + + handler = ResponsePollingHandler(redis_cache=mock_redis, ttl=3600) + + await handler.update_state( + polling_id="litellm_poll_test", + status="in_progress", + ) + + # Verify the update was saved + mock_redis.async_set_cache.assert_called_once() + call_args = mock_redis.async_set_cache.call_args + stored = json.loads(call_args.kwargs["value"]) + + assert stored["status"] == "in_progress" + + @pytest.mark.asyncio + async def test_update_state_replaces_full_output_list(self): + """Test that update_state replaces the full output list""" + mock_redis = AsyncMock() + mock_redis.async_get_cache.return_value = json.dumps({ + "id": "litellm_poll_test", + "object": "response", + "status": "in_progress", + "output": [{"id": "old_item", "type": "message"}], + "created_at": 1234567890 + }) + + handler = ResponsePollingHandler(redis_cache=mock_redis, ttl=3600) + + new_output = [ + {"id": "item_1", "type": "message", "content": [{"type": "text", "text": "Hello"}]}, + {"id": "item_2", "type": "message", "content": [{"type": "text", "text": "World"}]}, + ] + + await handler.update_state( + polling_id="litellm_poll_test", + output=new_output, + ) + + call_args = mock_redis.async_set_cache.call_args + stored = json.loads(call_args.kwargs["value"]) + + assert len(stored["output"]) == 2 + assert stored["output"][0]["id"] == "item_1" + assert stored["output"][1]["id"] == "item_2" + + @pytest.mark.asyncio + async def test_update_state_with_usage(self): + """Test that update_state correctly stores usage data""" + mock_redis = AsyncMock() + mock_redis.async_get_cache.return_value = json.dumps({ + "id": "litellm_poll_test", + "object": "response", + "status": "in_progress", + "output": [], + "created_at": 1234567890 + }) + + handler = ResponsePollingHandler(redis_cache=mock_redis) + + usage_data = { + "input_tokens": 10, + "output_tokens": 50, + "total_tokens": 60 + } + + await handler.update_state( + polling_id="litellm_poll_test", + status="completed", + usage=usage_data, + ) + + call_args = mock_redis.async_set_cache.call_args + stored = json.loads(call_args.kwargs["value"]) + + assert stored["status"] == "completed" + assert stored["usage"] == usage_data + + @pytest.mark.asyncio + async def test_update_state_with_reasoning_tools_tool_choice(self): + """Test that update_state stores reasoning, tools, and tool_choice from response.completed""" + mock_redis = AsyncMock() + mock_redis.async_get_cache.return_value = json.dumps({ + "id": "litellm_poll_test", + "object": "response", + "status": "in_progress", + "output": [], + "created_at": 1234567890 + }) + + handler = ResponsePollingHandler(redis_cache=mock_redis) + + reasoning_data = {"effort": "medium", "summary": "Step by step analysis"} + tool_choice_data = {"type": "function", "function": {"name": "get_weather"}} + tools_data = [{"type": "function", "function": {"name": "get_weather", "parameters": {}}}] + + await handler.update_state( + polling_id="litellm_poll_test", + status="completed", + reasoning=reasoning_data, + tool_choice=tool_choice_data, + tools=tools_data, + ) + + call_args = mock_redis.async_set_cache.call_args + stored = json.loads(call_args.kwargs["value"]) + + assert stored["reasoning"] == reasoning_data + assert stored["tool_choice"] == tool_choice_data + assert stored["tools"] == tools_data + + @pytest.mark.asyncio + async def test_update_state_with_error_sets_failed_status(self): + """Test that providing an error automatically sets status to failed""" + mock_redis = AsyncMock() + mock_redis.async_get_cache.return_value = json.dumps({ + "id": "litellm_poll_test", + "object": "response", + "status": "in_progress", + "output": [], + "created_at": 1234567890 + }) + + handler = ResponsePollingHandler(redis_cache=mock_redis) + + error_data = { + "type": "internal_error", + "message": "Something went wrong", + "code": "server_error" + } + + await handler.update_state( + polling_id="litellm_poll_test", + error=error_data, + ) + + call_args = mock_redis.async_set_cache.call_args + stored = json.loads(call_args.kwargs["value"]) + + assert stored["status"] == "failed" + assert stored["error"] == error_data + + @pytest.mark.asyncio + async def test_update_state_with_incomplete_details(self): + """Test that update_state stores incomplete_details""" + mock_redis = AsyncMock() + mock_redis.async_get_cache.return_value = json.dumps({ + "id": "litellm_poll_test", + "object": "response", + "status": "in_progress", + "output": [], + "created_at": 1234567890 + }) + + handler = ResponsePollingHandler(redis_cache=mock_redis) + + incomplete_details = { + "reason": "max_output_tokens" + } + + await handler.update_state( + polling_id="litellm_poll_test", + status="incomplete", + incomplete_details=incomplete_details, + ) + + call_args = mock_redis.async_set_cache.call_args + stored = json.loads(call_args.kwargs["value"]) + + assert stored["status"] == "incomplete" + assert stored["incomplete_details"] == incomplete_details + + @pytest.mark.asyncio + async def test_update_state_does_nothing_without_redis(self): + """Test that update_state gracefully handles no Redis cache""" + handler = ResponsePollingHandler(redis_cache=None) + + # Should not raise an exception + await handler.update_state( + polling_id="litellm_poll_test", + status="in_progress", + ) + + @pytest.mark.asyncio + async def test_update_state_handles_missing_cached_state(self): + """Test that update_state handles case when cached state doesn't exist""" + mock_redis = AsyncMock() + mock_redis.async_get_cache.return_value = None # Cache miss + + handler = ResponsePollingHandler(redis_cache=mock_redis) + + # Should not raise an exception + await handler.update_state( + polling_id="litellm_poll_test", + status="in_progress", + ) + + # Should not try to set cache if nothing was found + mock_redis.async_set_cache.assert_not_called() + + # ==================== Get State Tests ==================== + + @pytest.mark.asyncio + async def test_get_state_returns_cached_state(self): + """Test that get_state returns the cached state""" + mock_redis = AsyncMock() + cached_state = { + "id": "litellm_poll_test", + "object": "response", + "status": "in_progress", + "output": [{"id": "item_1", "type": "message"}], + "created_at": 1234567890, + "usage": {"input_tokens": 10, "output_tokens": 20} + } + mock_redis.async_get_cache.return_value = json.dumps(cached_state) + + handler = ResponsePollingHandler(redis_cache=mock_redis) + + result = await handler.get_state("litellm_poll_test") + + assert result == cached_state + + @pytest.mark.asyncio + async def test_get_state_returns_none_for_missing_state(self): + """Test that get_state returns None when state doesn't exist""" + mock_redis = AsyncMock() + mock_redis.async_get_cache.return_value = None + + handler = ResponsePollingHandler(redis_cache=mock_redis) + + result = await handler.get_state("litellm_poll_nonexistent") + + assert result is None + + @pytest.mark.asyncio + async def test_get_state_returns_none_without_redis(self): + """Test that get_state returns None when Redis is not configured""" + handler = ResponsePollingHandler(redis_cache=None) + + result = await handler.get_state("litellm_poll_test") + + assert result is None + + # ==================== Cancel Polling Tests ==================== + + @pytest.mark.asyncio + async def test_cancel_polling_updates_status_to_cancelled(self): + """Test that cancel_polling sets status to cancelled""" + mock_redis = AsyncMock() + mock_redis.async_get_cache.return_value = json.dumps({ + "id": "litellm_poll_test", + "object": "response", + "status": "in_progress", + "output": [], + "created_at": 1234567890 + }) + + handler = ResponsePollingHandler(redis_cache=mock_redis) + + result = await handler.cancel_polling("litellm_poll_test") + + assert result is True + + call_args = mock_redis.async_set_cache.call_args + stored = json.loads(call_args.kwargs["value"]) + assert stored["status"] == "cancelled" + + # ==================== Delete Polling Tests ==================== + + @pytest.mark.asyncio + async def test_delete_polling_removes_from_cache(self): + """Test that delete_polling removes the entry from Redis""" + mock_redis = AsyncMock() + mock_async_client = AsyncMock() + mock_redis.redis_async_client = True # hasattr check + mock_redis.init_async_client.return_value = mock_async_client + + handler = ResponsePollingHandler(redis_cache=mock_redis) + + result = await handler.delete_polling("litellm_poll_test") + + assert result is True + mock_async_client.delete.assert_called_once_with( + "litellm:polling:response:litellm_poll_test" + ) + + @pytest.mark.asyncio + async def test_delete_polling_returns_false_without_redis(self): + """Test that delete_polling returns False when Redis is not configured""" + handler = ResponsePollingHandler(redis_cache=None) + + result = await handler.delete_polling("litellm_poll_test") + + assert result is False + + # ==================== TTL Tests ==================== + + def test_default_ttl_is_one_hour(self): + """Test that default TTL is 3600 seconds (1 hour)""" + handler = ResponsePollingHandler(redis_cache=None) + + assert handler.ttl == 3600 + + def test_custom_ttl_is_respected(self): + """Test that custom TTL is stored correctly""" + handler = ResponsePollingHandler(redis_cache=None, ttl=7200) + + assert handler.ttl == 7200 + + @pytest.mark.asyncio + async def test_update_state_uses_configured_ttl(self): + """Test that update_state uses the configured TTL""" + mock_redis = AsyncMock() + mock_redis.async_get_cache.return_value = json.dumps({ + "id": "litellm_poll_test", + "object": "response", + "status": "queued", + "output": [], + "created_at": 1234567890 + }) + + handler = ResponsePollingHandler(redis_cache=mock_redis, ttl=1800) + + await handler.update_state( + polling_id="litellm_poll_test", + status="in_progress", + ) + + call_args = mock_redis.async_set_cache.call_args + assert call_args.kwargs["ttl"] == 1800 + + +class TestStreamingEventProcessing: + """ + Test cases for streaming event processing logic. + + These tests verify the expected behavior when processing different + OpenAI streaming event types. + """ + + def test_accumulated_text_structure(self): + """Test the structure used for accumulating text deltas""" + accumulated_text = {} + + # Simulate accumulating deltas for (item_id, content_index) + key = ("item_123", 0) + accumulated_text[key] = "" + accumulated_text[key] += "Hello " + accumulated_text[key] += "World" + + assert accumulated_text[key] == "Hello World" + assert ("item_123", 0) in accumulated_text + assert ("item_123", 1) not in accumulated_text + + def test_output_items_tracking_structure(self): + """Test the structure used for tracking output items by ID""" + output_items = {} + + # Simulate adding output items + item1 = {"id": "item_1", "type": "message", "content": []} + item2 = {"id": "item_2", "type": "function_call", "name": "get_weather"} + + output_items[item1["id"]] = item1 + output_items[item2["id"]] = item2 + + assert len(output_items) == 2 + assert output_items["item_1"]["type"] == "message" + assert output_items["item_2"]["type"] == "function_call" + + def test_150ms_batch_interval_constant(self): + """Test that the batch interval is 150ms""" + UPDATE_INTERVAL = 0.150 # 150ms + + assert UPDATE_INTERVAL == 0.150 + assert UPDATE_INTERVAL * 1000 == 150 # 150 milliseconds +