Merge pull request temporal-sa#5 from robholland/rh-redis-refactor

Refactor redis architecture.

Author: rross

README.md

@@ -23,7 +23,7 @@ You can run through the scenarios with the Temporal version using a [Web Applica
 * [Poetry](https://python-poetry.org/docs/) - Python Dependency Management
 * [OpenAI API Key] (https://platform.openai.com/api-keys) - Your key to accessing OpenAI's LLM
 * [Temporal CLI](https://docs.temporal.io/cli#install) - Local Temporal service
-* [Redis](https://redis.io/downloads/) - Workflow writes conversation history, API reads from it
+* [Redis](https://redis.io/downloads/) - Stores conversation history and real-time status updates
 
 ## Set up Python Environment
 ```bash
@@ -43,5 +43,7 @@ It should look something like this:
 export OPENAI_API_KEY=sk-proj-....
 ```
 
+## Getting Started
+
 See the OpenAI Agents SDK Version [here](src/oai_supervisor/README.md)
 And the Temporal version of this example is located [here](src/temporal_supervisor/README.md)

images/application-architecture.png

@@ -16,7 +16,7 @@ dependencies = [
     "redis (>=6.2.0,<7.0.0)",
     "aiohttp (>=3.12.14,<4.0.0)",
     "openai-agents (>=0.2.3,<0.3.0)",
-    "temporalio (>=1.15.0,<2.0.0)",
+    "temporalio (>=1.17.0,<2.0.0)",
 ]
 
 [tool.poetry]

poetry.lock

@@ -1,20 +1,16 @@
-import json
-
-import asyncio
 from contextlib import asynccontextmanager
 from typing import Optional, AsyncGenerator
 
-from fastapi import FastAPI, HTTPException, Request
+from fastapi import FastAPI, HTTPException, Request, Query
 from fastapi.middleware.cors import CORSMiddleware
-from fastapi.responses import StreamingResponse
-from temporalio.client import Client, WithStartWorkflowOperation
-from temporalio.common import WorkflowIDReusePolicy, WorkflowIDConflictPolicy
+from temporalio.client import Client
+from temporalio.common import WorkflowIDReusePolicy
 from temporalio.exceptions import TemporalError
 from temporalio.contrib.openai_agents import OpenAIAgentsPlugin
 from temporalio.service import RPCError
 
+from common.event_stream_manager import EventStreamManager
 from common.client_helper import ClientHelper
-from common.db_manager import DBManager
 from common.user_message import ProcessUserMessageInput
 from temporal_supervisor.claim_check.claim_check_plugin import ClaimCheckPlugin
 from temporal_supervisor.workflows.supervisor_workflow import WealthManagementWorkflow
@@ -59,10 +55,12 @@ def root():
     return {"message": "OpenAI Agent SDK + Temporal Agent!"}
 
 @app.get("/get-chat-history")
-async def get_chat_history():
+async def get_chat_history(
+    from_index: int = Query(0, description="Get events starting from this index")
+):
     """ Retrieves the chat history from Redis """
     try:
-        history = await DBManager().read(WORKFLOW_ID)
+        history = await EventStreamManager().get_events_from_index(WORKFLOW_ID, from_index)
         if history is None:
             return ""
 
@@ -80,14 +78,6 @@ async def get_chat_history():
 
 @app.post("/send-prompt")
 async def send_prompt(prompt: str):
-    # Start or update the workflow
-    start_op = WithStartWorkflowOperation(
-        WealthManagementWorkflow.run,
-        id=WORKFLOW_ID,
-        task_queue=task_queue,
-        id_conflict_policy=WorkflowIDConflictPolicy.USE_EXISTING,
-    )
-
     print(f"Received prompt {prompt}")
 
     message = ProcessUserMessageInput(
@@ -123,11 +113,10 @@ async def end_chat():
 @app.post("/start-workflow")
 async def start_workflow(request: Request):
     try:
-        sse_url = str(request.url_for(UPDATE_STATUS_NAME))
         # start the workflow
         await temporal_client.start_workflow(
             WealthManagementWorkflow.run,
-            args=[sse_url],
+            args=[],
             id=WORKFLOW_ID,
             task_queue=task_queue,
             id_reuse_policy=WorkflowIDReusePolicy.ALLOW_DUPLICATE
@@ -141,33 +130,3 @@ async def start_workflow(request: Request):
         return {
             "message": f"An error occurred starting the workflow {e}"
         }
-
-# In-memory list to hold active SSE client connections
-# Note that this does not scale past one instance of the API
-connected_clients = []
-
-# SSE generator function
-async def event_generator(request: Request):
-    client_queue = asyncio.Queue()
-    connected_clients.append(client_queue)
-    try:
-        while True:
-            # Wait for a new message to be put in the queue
-            message = await client_queue.get()
-            yield f"data: {message}\n\n"
-    except asyncio.CancelledError:
-        connected_clients.remove(client_queue)
-        raise
-
-# Endpoint for clients to connect and receive events
-@app.get("/sse/status/stream")
-async def sse_endpoint(request: Request):
-    return StreamingResponse(event_generator(request), media_type="text/event-stream")
-
-# Endpoint for the Temporal Workflow to send updates
-@app.post("/sse/status/update", name=UPDATE_STATUS_NAME)
-async def update_status(data: dict):
-    message = json.dumps(data)
-    for queue in connected_clients:
-        await queue.put(message)
-    return {"message": "Status updated"}

pyproject.toml

@@ -0,0 +1,178 @@
+import json
+import time
+import os
+from typing import List, Dict, Any, Union
+from enum import Enum
+from dataclasses import asdict
+
+import redis.asyncio as redis
+
+from .user_message import ChatInteraction
+from .status_update import StatusUpdate
+
+class EventType(str, Enum):
+    """Event types for the chat event stream"""
+    CHAT_INTERACTION = "chat_interaction"
+    STATUS_UPDATE = "status_update"
+
+class EventStreamManager:
+    """
+    Manages a Redis list-based event stream for chat conversations.
+    
+    Uses LPUSH for O(1) appends and LRANGE for efficient range queries.
+    Each event has an implicit sequence number based on its position in the list.
+    """
+    
+    def __init__(self, redis_host: str = None, redis_port: int = None):
+        self.redis_host = redis_host or os.getenv("REDIS_HOST", "localhost")
+        self.redis_port = redis_port or int(os.getenv("REDIS_PORT", "6379"))
+        self.redis_client = redis.Redis(
+            host=self.redis_host, 
+            port=self.redis_port,
+            decode_responses=True
+        )
+    
+    def _get_stream_key(self, workflow_id: str) -> str:
+        """Get the Redis key for the event stream"""
+        return f"events:{workflow_id}"
+    
+    def _get_meta_key(self, workflow_id: str) -> str:
+        """Get the Redis key for stream metadata"""
+        return f"events:{workflow_id}:meta"
+        
+    async def append_chat_interaction(
+        self, 
+        workflow_id: str, 
+        chat_interaction: ChatInteraction
+    ) -> int:
+        """
+        Append a chat interaction to the stream.
+        
+        Returns the new total length of the event stream.
+        """
+        return await self._append_domain_event(
+            workflow_id, 
+            EventType.CHAT_INTERACTION, 
+            chat_interaction
+        )
+    
+    async def append_status_update(
+        self, 
+        workflow_id: str, 
+        status_update: StatusUpdate
+    ) -> int:
+        """
+        Append a status update to the stream.
+        
+        Returns the new total length of the event stream.
+        """
+        return await self._append_domain_event(
+            workflow_id, 
+            EventType.STATUS_UPDATE, 
+            status_update
+        )
+    
+    async def _append_domain_event(
+        self, 
+        workflow_id: str, 
+        event_type: EventType, 
+        domain_object: Union[ChatInteraction, StatusUpdate]
+    ) -> int:
+        """
+        Internal method to append domain objects to the stream.
+        
+        Returns the new total length of the event stream.
+        """
+        stream_key = self._get_stream_key(workflow_id)
+        
+        # Convert domain object to dict
+        content_dict = asdict(domain_object)
+        
+        # Build the event with structured content
+        event = {
+            "type": event_type.value,
+            "content": content_dict
+        }
+        
+        event_json = json.dumps(event)
+        
+        # Use RPUSH to add to the end (chronological order)
+        # RPUSH returns the new length of the list after insertion
+        new_length = await self.redis_client.rpush(stream_key, event_json)
+        
+        return new_length
+    
+    async def get_events_from_index(
+        self, 
+        workflow_id: str, 
+        from_index: int = 0
+    ) -> List[Dict[str, Any]]:
+        """
+        Get events starting from a specific index.
+        
+        Args:
+            workflow_id: The workflow ID
+            from_index: Start from this index (0-based)
+            
+        Returns:
+            List of events in chronological order
+        """
+        stream_key = self._get_stream_key(workflow_id)
+        
+        # Get all events from the specified index to the end
+        event_strings = await self.redis_client.lrange(stream_key, from_index, -1)
+        
+        # Parse events
+        events = []
+        for event_str in event_strings:
+            try:
+                event = json.loads(event_str)
+                events.append(event)
+            except json.JSONDecodeError:
+                continue  # Skip malformed events
+        
+        return events
+    
+    async def get_all_events(self, workflow_id: str) -> List[Dict[str, Any]]:
+        """
+        Get all events in the stream.
+        
+        Returns events in chronological order.
+        """
+        stream_key = self._get_stream_key(workflow_id)
+        
+        # Get all events
+        event_strings = await self.redis_client.lrange(stream_key, 0, -1)
+        
+        # Parse events (already in chronological order due to RPUSH)
+        events = []
+        for event_str in event_strings:
+            try:
+                events.append(json.loads(event_str))
+            except json.JSONDecodeError:
+                continue  # Skip malformed events
+        
+        return events
+    
+    async def get_total_events(self, workflow_id: str) -> int:
+        """Get the total number of events in the stream"""
+        stream_key = self._get_stream_key(workflow_id)
+        return await self.redis_client.llen(stream_key)
+    
+    async def delete_stream(self, workflow_id: str) -> bool:
+        """
+        Delete the entire event stream for a workflow.
+        
+        Returns True if the stream was deleted, False if it didn't exist.
+        """
+        stream_key = self._get_stream_key(workflow_id)
+        meta_key = self._get_meta_key(workflow_id)
+        
+        # Delete both the stream and metadata
+        deleted = await self.redis_client.delete(stream_key, meta_key)
+        
+        return deleted > 0
+    
+    async def close(self):
+        """Close the Redis connection"""
+        await self.redis_client.aclose()

src/api/main.py

@@ -0,0 +1,8 @@
+from dataclasses import dataclass
+
+@dataclass
+class StatusUpdate:
+    status: str
+
+    def __str__(self):
+        return f"Status: {self.status}"