Port main branch updates to refactored api/common/conserver structure

Port main branch updates to refactored api/common/conserver structure

Author: pavanputhra

.gitignore

@@ -21,4 +21,5 @@ tmp
 .qodo
 
 traefik/
-redis_data/
+redis_data/
+litellm_config.yaml

api/api.py

@@ -954,7 +954,7 @@ async def get_vcon_count(
 async def get_config() -> JSONResponse:
     """Get the current system configuration.
 
-    Reads and returns the configuration from the file specified in CONSERVER_CONFIG_FILE.
+    Returns the current configuration via Configuration.get_config().
 
     Returns:
         JSONResponse containing the configuration
@@ -963,8 +963,7 @@ async def get_config() -> JSONResponse:
         HTTPException: If there is an error reading the config file
     """
     try:
-        with open(os.getenv("CONSERVER_CONFIG_FILE"), "r") as f:
-            config = yaml.safe_load(f)
+        config = Configuration.get_config()
         return JSONResponse(content=config)
     except Exception as e:
         logger.error(f"Error reading config: {str(e)}")

common/lib/openai_client.py

@@ -0,0 +1,132 @@
+"""
+Shared OpenAI/Azure/LiteLLM client for vcon-server.
+
+When LITELLM_PROXY_URL and LITELLM_MASTER_KEY are set in opts,
+returns an OpenAI client configured to use the LiteLLM proxy. Otherwise uses
+direct OpenAI or Azure OpenAI credentials from opts.
+
+All links and storage that call OpenAI should use get_openai_client(opts) so
+LLM provider and proxy can be switched in one place.
+"""
+from openai import OpenAI, AzureOpenAI, AsyncOpenAI, AsyncAzureOpenAI
+
+from lib.logging_utils import init_logger
+
+logger = init_logger(__name__)
+
+# Default Azure API version when not specified
+DEFAULT_AZURE_OPENAI_API_VERSION = "2024-10-21"
+
+
+def get_openai_client(opts=None):
+    """
+    Return an OpenAI-compatible client (OpenAI or AzureOpenAI).
+    Same client is used for chat and embeddings; LiteLLM proxy supports both.
+
+    opts: dict of options. All values are read from opts only.
+
+    Supported keys in opts:
+      - LITELLM_PROXY_URL, LITELLM_MASTER_KEY  -> use LiteLLM proxy (chat + embeddings)
+      - AZURE_OPENAI_ENDPOINT, AZURE_OPENAI_API_KEY, AZURE_OPENAI_API_VERSION -> Azure
+      - OPENAI_API_KEY or openai_api_key or api_key -> OpenAI
+      - organization / organization_key, project / project_key (optional)
+    """
+    opts = opts or {}
+
+    litellm_url = (opts.get("LITELLM_PROXY_URL") or "").strip().rstrip("/")
+    litellm_key = (opts.get("LITELLM_MASTER_KEY") or "").strip()
+
+    if litellm_url and litellm_key:
+        logger.info("Using LiteLLM proxy at %s", litellm_url)
+        organization = opts.get("organization") or opts.get("organization_key")
+        project = opts.get("project") or opts.get("project_key")
+        return OpenAI(
+            api_key=litellm_key,
+            base_url=litellm_url,
+            organization=organization if organization else None,
+            project=project if project else None,
+            timeout=120.0,
+            max_retries=0,
+        )
+
+    azure_endpoint = (opts.get("AZURE_OPENAI_ENDPOINT") or "").strip()
+    azure_api_key = (opts.get("AZURE_OPENAI_API_KEY") or "").strip()
+    azure_api_version = opts.get("AZURE_OPENAI_API_VERSION") or DEFAULT_AZURE_OPENAI_API_VERSION
+
+    if azure_endpoint and azure_api_key:
+        logger.info("Using Azure OpenAI client at endpoint: %s", azure_endpoint)
+        return AzureOpenAI(
+            api_key=azure_api_key,
+            azure_endpoint=azure_endpoint,
+            api_version=azure_api_version,
+            timeout=120.0,
+            max_retries=0,
+        )
+
+    openai_api_key = (
+        opts.get("OPENAI_API_KEY")
+        or opts.get("openai_api_key")
+        or opts.get("api_key")
+    )
+    if openai_api_key:
+        logger.info("Using public OpenAI client")
+        organization = opts.get("organization") or opts.get("organization_key")
+        project = opts.get("project") or opts.get("project_key")
+        return OpenAI(
+            api_key=openai_api_key,
+            organization=organization if organization else None,
+            project=project if project else None,
+            timeout=120.0,
+            max_retries=0,
+        )
+
+    raise ValueError(
+        "Set LITELLM_PROXY_URL + LITELLM_MASTER_KEY, or "
+        "AZURE_OPENAI_ENDPOINT + AZURE_OPENAI_API_KEY, or OPENAI_API_KEY (or api_key)"
+    )
+
+
+def get_async_openai_client(opts=None):
+    """
+    Return an async OpenAI-compatible client. Same opts semantics as get_openai_client.
+    LiteLLM proxy is used for both chat and embeddings when configured.
+    """
+    opts = opts or {}
+
+    litellm_url = (opts.get("LITELLM_PROXY_URL") or "").strip().rstrip("/")
+    litellm_key = (opts.get("LITELLM_MASTER_KEY") or "").strip()
+    if litellm_url and litellm_key:
+        logger.info("Using LiteLLM proxy at %s (async)", litellm_url)
+        return AsyncOpenAI(
+            api_key=litellm_key,
+            base_url=litellm_url + "/v1",
+            timeout=120.0,
+            max_retries=0,
+        )
+
+    azure_endpoint = (opts.get("AZURE_OPENAI_ENDPOINT") or "").strip()
+    azure_api_key = (opts.get("AZURE_OPENAI_API_KEY") or "").strip()
+    azure_api_version = opts.get("AZURE_OPENAI_API_VERSION") or DEFAULT_AZURE_OPENAI_API_VERSION
+    if azure_endpoint and azure_api_key:
+        logger.info("Using Azure OpenAI client at endpoint: %s (async)", azure_endpoint)
+        return AsyncAzureOpenAI(
+            api_key=azure_api_key,
+            azure_endpoint=azure_endpoint,
+            api_version=azure_api_version,
+            timeout=120.0,
+            max_retries=0,
+        )
+
+    openai_api_key = (
+        opts.get("OPENAI_API_KEY")
+        or opts.get("openai_api_key")
+        or opts.get("api_key")
+    )
+    if openai_api_key:
+        logger.info("Using public OpenAI client (async)")
+        return AsyncOpenAI(api_key=openai_api_key, timeout=120.0, max_retries=0)
+
+    raise ValueError(
+        "Set LITELLM_PROXY_URL + LITELLM_MASTER_KEY, or "
+        "AZURE_OPENAI_ENDPOINT + AZURE_OPENAI_API_KEY, or OPENAI_API_KEY (or api_key)"
+    )

common/storage/chatgpt_files/__init__.py

@@ -1,8 +1,8 @@
 from lib.logging_utils import init_logger
+from lib.openai_client import get_openai_client
 import json
 import os
 import redis_mgr
-from openai import OpenAI
 
 logger = init_logger(__name__)
 
@@ -29,12 +29,9 @@ def save(vcon_uuid: str, options: dict = default_options) -> None:
         file_name = f"{vcon_uuid}.vcon.json"
         with open(file_name, "w") as file:
             json.dump(vcon, file)
-        client = OpenAI(
-            organization=options["organization_key"],
-            project=options["project_key"],
-            api_key=options["api_key"],
-        )
-        file = client.files.create(file=open(file_name, "rb"), purpose=options["purpose"])
+        client = get_openai_client(options)
+        with open(file_name, "rb") as upload_file:
+            file = client.files.create(file=upload_file, purpose=options["purpose"])
         os.remove(file_name)
         client.beta.vector_stores.files.create(vector_store_id=options["vector_store_id"], file_id=file.id)
     except Exception as error:

common/storage/milvus/__init__.py

@@ -14,7 +14,7 @@
 
 try:
     from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType, utility
-    from openai import OpenAI
+    from lib.openai_client import get_openai_client
 except ImportError:
     logging.error("Required packages not found. Install with: pip install pymilvus openai")
     raise
@@ -446,12 +446,9 @@ def save(vcon_uuid: str, opts=default_options) -> None:
                 logger.info(f"vCon {vcon_uuid} already exists in Milvus collection {collection_name}, skipping")
                 return
 
-        # Initialize OpenAI client
-        openai_client = OpenAI(
-            api_key=opts["api_key"],
-            organization=opts["organization"] if opts["organization"] else None
-        )
-        
+        # Initialize OpenAI client (supports LiteLLM proxy via LITELLM_PROXY_URL + LITELLM_MASTER_KEY provided in opts)
+        openai_client = get_openai_client(opts)
+
         # Extract text content from vCon
         text = extract_text_from_vcon(vcon_dict)
 

common/tests/test_external_ingress.py

@@ -39,9 +39,9 @@ def test_validate_ingress_api_key_function(self):
 
     @patch("config.Configuration.get_ingress_auth")
     @patch("api.add_vcon_to_set")
-    @patch("api.index_vcon_parties")
+    @patch("api.index_vcon")
     def test_successful_submission_single_api_key(
-        self, mock_index_vcon_parties, mock_add_vcon_to_set, mock_get_ingress_auth
+        self, mock_index_vcon, mock_add_vcon_to_set, mock_get_ingress_auth
     ):
         """Test successful vCon submission with single API key configuration."""
         # Configure mocks
@@ -89,9 +89,9 @@ def test_successful_submission_single_api_key(
 
     @patch("config.Configuration.get_ingress_auth")
     @patch("api.add_vcon_to_set")
-    @patch("api.index_vcon_parties")
+    @patch("api.index_vcon")
     def test_successful_submission_multiple_api_keys(
-        self, mock_index_vcon_parties, mock_add_vcon_to_set, mock_get_ingress_auth
+        self, mock_index_vcon, mock_add_vcon_to_set, mock_get_ingress_auth
     ):
         """Test successful vCon submission with multiple API keys for same ingress."""
         # Configure mocks - multiple API keys for same ingress list

common/tests/test_post_vcon_expiry.py

@@ -69,6 +69,7 @@ def test_post_vcon_stores_without_default_expiry(
 
     def test_post_vcon_expiry_value_is_3600(self):
         """Test that the default expiry value is 3600 seconds (1 hour)."""
+        # Verify the configured value
         assert VCON_REDIS_EXPIRY == 3600, "Default VCON_REDIS_EXPIRY should be 3600 seconds"
 
     @patch("api.add_vcon_to_set")

conserver/links/analyze/__init__.py

@@ -1,7 +1,7 @@
 from lib.vcon_redis import VconRedis
 from lib.logging_utils import init_logger
+from lib.openai_client import get_openai_client
 import logging
-from openai import OpenAI, AzureOpenAI
 from tenacity import (
     retry,
     stop_after_attempt,
@@ -98,24 +98,7 @@ def run(
         logger.info(f"Skipping {link_name} vCon {vcon_uuid} due to sampling")
         return vcon_uuid
 
-    # Extract credentials from options
-    openai_api_key = opts.get("OPENAI_API_KEY")
-    azure_openai_api_key = opts.get("AZURE_OPENAI_API_KEY")
-    azure_openai_endpoint = opts.get("AZURE_OPENAI_ENDPOINT")
-    api_version = opts.get("AZURE_OPENAI_API_VERSION")
-
-    client = None
-    if openai_api_key:
-        client = OpenAI(api_key=openai_api_key, timeout=120.0, max_retries=0)
-        logger.info("Using public OpenAI client")
-    elif azure_openai_api_key and azure_openai_endpoint:
-        client = AzureOpenAI(api_key=azure_openai_api_key, azure_endpoint=azure_openai_endpoint, api_version=api_version)
-        logger.info(f"Using Azure OpenAI client at endpoint:{azure_openai_endpoint}")
-    else:
-        raise ValueError(
-            "OpenAI or Azure OpenAI credentials not provided. "
-            "Need OPENAI_API_KEY or AZURE_OPENAI_API_KEY and AZURE_OPENAI_ENDPOINT"
-        )
+    client = get_openai_client(opts)
 
     source_type = navigate_dict(opts, "source.analysis_type")
     text_location = navigate_dict(opts, "source.text_location")

conserver/links/analyze_and_label/__init__.py

@@ -1,8 +1,8 @@
 from lib.vcon_redis import VconRedis
 from lib.logging_utils import init_logger
+from lib.openai_client import get_openai_client
 import logging
 import json
-from openai import OpenAI
 from tenacity import (
     retry,
     stop_after_attempt,
@@ -79,7 +79,7 @@ def run(
         logger.info(f"Skipping {link_name} vCon {vcon_uuid} due to sampling")
         return vcon_uuid
 
-    client = OpenAI(api_key=opts["OPENAI_API_KEY"], timeout=120.0, max_retries=0)
+    client = get_openai_client(opts)
     source_type = navigate_dict(opts, "source.analysis_type")
     text_location = navigate_dict(opts, "source.text_location")
 

conserver/links/analyze_and_label/tests/test_analyze_and_label.py

@@ -1,7 +1,7 @@
 import os
 import json
 import pytest
-from unittest.mock import patch, MagicMock
+from unittest.mock import patch, MagicMock, Mock
 
 from server.links.analyze_and_label import run, generate_analysis_with_labels, get_analysis_for_type, navigate_dict
 from server.vcon import Vcon
@@ -222,11 +222,13 @@ def test_navigate_dict():
     assert navigate_dict(test_dict, "z") is None
 
 
+@patch('server.links.analyze_and_label.get_openai_client')
 @patch('server.links.analyze_and_label.generate_analysis_with_labels')
 @patch('server.links.analyze_and_label.is_included', return_value=True)
 @patch('server.links.analyze_and_label.randomly_execute_with_sampling', return_value=True)
-def test_run_basic(mock_sampling, mock_is_included, mock_generate_analysis, mock_redis_with_vcon, sample_vcon):
+def test_run_basic(mock_sampling, mock_is_included, mock_generate_analysis, mock_get_client, mock_redis_with_vcon, sample_vcon):
     """Test the basic run functionality with mocked analysis generation"""
+    mock_get_client.return_value = Mock()
     # Set up mock to return analysis JSON
     mock_generate_analysis.return_value = json.dumps({
         "labels": ["customer_service", "billing_issue", "refund"]
@@ -264,12 +266,14 @@ def test_run_basic(mock_sampling, mock_is_included, mock_generate_analysis, mock
     assert "refund:refund" in tags_attachment["body"]
 
 
+@patch('server.links.analyze_and_label.get_openai_client')
 @patch('server.links.analyze_and_label.get_analysis_for_type')
 @patch('server.links.analyze_and_label.generate_analysis_with_labels')
 @patch('server.links.analyze_and_label.is_included', return_value=True)
 @patch('server.links.analyze_and_label.randomly_execute_with_sampling', return_value=True)
-def test_run_skip_existing_analysis(mock_sampling, mock_is_included, mock_generate_analysis, mock_get_analysis, mock_redis_with_vcon, sample_vcon_with_analysis):
+def test_run_skip_existing_analysis(mock_sampling, mock_is_included, mock_generate_analysis, mock_get_analysis, mock_get_client, mock_redis_with_vcon, sample_vcon_with_analysis):
     """Test that run skips dialogs with existing labeled analysis"""
+    mock_get_client.return_value = Mock()
     # Set up mock for generate_analysis_with_labels
     mock_generate_analysis.return_value = json.dumps({
         "labels": ["new_label_that_should_not_be_added"]
@@ -307,11 +311,13 @@ def test_run_skip_existing_analysis(mock_sampling, mock_is_included, mock_genera
     mock_generate_analysis.assert_not_called()
 
 
+@patch('server.links.analyze_and_label.get_openai_client')
 @patch('server.links.analyze_and_label.generate_analysis_with_labels')
 @patch('server.links.analyze_and_label.is_included', return_value=True)
 @patch('server.links.analyze_and_label.randomly_execute_with_sampling', return_value=True)
-def test_run_json_parse_error(mock_sampling, mock_is_included, mock_generate_analysis, mock_redis_with_vcon, sample_vcon):
+def test_run_json_parse_error(mock_sampling, mock_is_included, mock_generate_analysis, mock_get_client, mock_redis_with_vcon, sample_vcon):
     """Test handling of JSON parse errors"""
+    mock_get_client.return_value = Mock()
     # Set up mock to return invalid JSON
     mock_generate_analysis.return_value = "This is not valid JSON"
 
@@ -338,11 +344,13 @@ def test_run_json_parse_error(mock_sampling, mock_is_included, mock_generate_ana
     assert tags_attachment is None or len(tags_attachment["body"]) == 0
 
 
+@patch('server.links.analyze_and_label.get_openai_client')
 @patch('server.links.analyze_and_label.generate_analysis_with_labels')
 @patch('server.links.analyze_and_label.is_included', return_value=True)
 @patch('server.links.analyze_and_label.randomly_execute_with_sampling', return_value=True)
-def test_run_analysis_exception(mock_sampling, mock_is_included, mock_generate_analysis, mock_redis_with_vcon, sample_vcon):
+def test_run_analysis_exception(mock_sampling, mock_is_included, mock_generate_analysis, mock_get_client, mock_redis_with_vcon, sample_vcon):
     """Test handling of analysis generation exceptions"""
+    mock_get_client.return_value = Mock()
     # Make analysis function raise an exception
     mock_generate_analysis.side_effect = Exception("Analysis generation failed")
 
@@ -358,11 +366,13 @@ def test_run_analysis_exception(mock_sampling, mock_is_included, mock_generate_a
         run("test-uuid", "analyze_and_label", opts)
 
 
+@patch('server.links.analyze_and_label.get_openai_client')
 @patch('server.links.analyze_and_label.generate_analysis_with_labels')
 @patch('server.links.analyze_and_label.is_included', return_value=True)
 @patch('server.links.analyze_and_label.randomly_execute_with_sampling', return_value=True)
-def test_run_message_format(mock_sampling, mock_is_included, mock_generate_analysis, mock_redis_with_vcon, sample_vcon_message_format):
+def test_run_message_format(mock_sampling, mock_is_included, mock_generate_analysis, mock_get_client, mock_redis_with_vcon, sample_vcon_message_format):
     """Test analyzing a dialog with message format"""
+    mock_get_client.return_value = Mock()
     # Set up the mock Redis instance to return our sample vCon with message format
     mock_instance = mock_redis_with_vcon.return_value
     mock_instance.get_vcon.return_value = sample_vcon_message_format