Revert "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/) - Stores conversation history and real-time status updates
+* [Redis](https://redis.io/downloads/) - Workflow writes conversation history, API reads from it
 
 ## Set up Python Environment
 ```bash
@@ -43,7 +43,5 @@ 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.17.0,<2.0.0)",
+    "temporalio (>=1.15.0,<2.0.0)",
 ]
 
 [tool.poetry]

poetry.lock

@@ -1,16 +1,20 @@
+import json
+
+import asyncio
 from contextlib import asynccontextmanager
 from typing import Optional, AsyncGenerator
 
-from fastapi import FastAPI, HTTPException, Request, Query
+from fastapi import FastAPI, HTTPException, Request
 from fastapi.middleware.cors import CORSMiddleware
-from temporalio.client import Client
-from temporalio.common import WorkflowIDReusePolicy
+from fastapi.responses import StreamingResponse
+from temporalio.client import Client, WithStartWorkflowOperation
+from temporalio.common import WorkflowIDReusePolicy, WorkflowIDConflictPolicy
 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
@@ -55,12 +59,10 @@ def root():
     return {"message": "OpenAI Agent SDK + Temporal Agent!"}
 
 @app.get("/get-chat-history")
-async def get_chat_history(
-    from_index: int = Query(0, description="Get events starting from this index")
-):
+async def get_chat_history():
     """ Retrieves the chat history from Redis """
     try:
-        history = await EventStreamManager().get_events_from_index(WORKFLOW_ID, from_index)
+        history = await DBManager().read(WORKFLOW_ID)
         if history is None:
             return ""
 
@@ -78,6 +80,14 @@ 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(
@@ -113,10 +123,11 @@ 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=[],
+            args=[sse_url],
             id=WORKFLOW_ID,
             task_queue=task_queue,
             id_reuse_policy=WorkflowIDReusePolicy.ALLOW_DUPLICATE
@@ -130,3 +141,33 @@ 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,28 @@
+import json
+
+import redis.asyncio as redis
+
+class DBManager:
+    def __init__(self, redis_host: str = "localhost", redis_port: int = 6379):
+        self.redis_client = redis.Redis(host=redis_host, port=redis_port)
+
+    # may need to revisit this later
+    async def save(self, key: str, value):
+        value_as_json = json.dumps(value)
+        await self.redis_client.set(key, value_as_json)
+
+    # may need to revisit this later
+    async def read(self, key: str) -> any:
+        print(f"Getting ready to retrieve the value for key {key}")
+        value =  await self.redis_client.get(key)
+        print(f"The value read is {value}")
+        if value is not None:
+            return_value = json.loads(value)
+            print(f"The return value after reading Redis and parsing json is {return_value}")
+            return return_value
+        # TODO: Validate this is okay
+        return None
+
+    async def delete(self, key: str):
+        print(f"Deleting key {key}")
+        await self.redis_client.delete(key)