Add concurrency

cohereAPI.py

@@ -1,10 +1,20 @@
 import cohere
 import asyncio
 
+# Global client variable for lazy initialization
+_client = None
+
+def get_client(api_key):
+    """Get or create Cohere client instance"""
+    global _client
+    if _client is None:
+        _client = cohere.ClientV2(api_key)
+    return _client
+
 def send_message_stream(system_message, user_message, conversation_history, api_key, model_name="command-a-03-2025"):
     """Stream response from Cohere API"""
-    # Initialize the Cohere client
-    co = cohere.ClientV2(api_key)
+    # Get or create the Cohere client
+    co = get_client(api_key)
     
     # Prepare all messages including history
     messages = [{"role": "system", "content": system_message}]
@@ -26,15 +36,46 @@ def send_message_stream(system_message, user_message, conversation_history, api_
             text_chunk = chunk.delta.message.content.text
             full_response += text_chunk
             yield text_chunk
+
+async def send_message_stream_async(system_message, user_message, conversation_history, api_key, model_name="command-a-03-2025"):
+    """Async wrapper for streaming response from Cohere API"""
+    def _sync_stream():
+        return send_message_stream(system_message, user_message, conversation_history, api_key, model_name)
+    
+    # Run the synchronous generator in a thread
+    loop = asyncio.get_event_loop()
+    
+    # Use a queue to handle the streaming data
+    queue = asyncio.Queue()
+    
+    def _stream_worker():
+        try:
+            for chunk in _sync_stream():
+                loop.call_soon_threadsafe(queue.put_nowait, chunk)
+        except Exception as e:
+            loop.call_soon_threadsafe(queue.put_nowait, StopIteration(e))
+        else:
+            loop.call_soon_threadsafe(queue.put_nowait, StopIteration())
+    
+    # Start the worker thread
+    import threading
+    thread = threading.Thread(target=_stream_worker)
+    thread.start()
+    
+    # Yield chunks asynchronously
+    while True:
+        chunk = await queue.get()
+        if isinstance(chunk, StopIteration):
+            if chunk.args:
+                raise chunk.args[0]
+            break
+        yield chunk
     
-    # Update conversation history after streaming is complete
-    conversation_history.append({"role": "user", "content": user_message})
-    conversation_history.append({"role": "assistant", "content": full_response})
 
 def send_message(system_message, user_message, conversation_history, api_key, model_name="command-a-03-2025"):
     """Non-streaming version for backward compatibility"""
-    # Initialize the Cohere client
-    co = cohere.ClientV2(api_key)
+    # Get or create the Cohere client
+    co = get_client(api_key)
     
     # Prepare all messages including history
     messages = [{"role": "system", "content": system_message}]
@@ -50,8 +91,15 @@ def send_message(system_message, user_message, conversation_history, api_key, mo
     # Get the response
     response_content = response.message.content[0].text
     
-    # Update conversation history for this session
-    conversation_history.append({"role": "user", "content": user_message})
-    conversation_history.append({"role": "assistant", "content": response_content})
-    
-    return response_content, conversation_history
+    return response_content
+
+async def send_message_async(system_message, user_message, conversation_history, api_key, model_name="command-a-03-2025"):
+    """Async version using asyncio.to_thread"""
+    return await asyncio.to_thread(
+        send_message, 
+        system_message, 
+        user_message, 
+        conversation_history, 
+        api_key, 
+        model_name
+    )