Merge pull request #1299 from solliancenet/kb-update-langchain-intera…

…ction-with-state-api

Update LangChainAPI interactions with the StateAPI

docs/development/development-local.md

@@ -337,11 +337,15 @@ The `CoreWorker` project is a .NET worker service that acts as the Cosmos DB cha
 
 ### Python Environment Variables
 
-Create a local environment variable named `FOUNDATIONALLM_APP_CONFIGURATION_URI`. The value should be the URI of the Azure App Configuration service and _not_ the connection string. We use role-based access controls (RBAC) to access the Azure App Configuration service, so the connection string is not required.
+Create local environment variables named:
+
+- `FOUNDATIONALLM_APP_CONFIGURATION_URI`. The value should be the URI of the Azure App Configuration service and _not_ the connection string. We use role-based access controls (RBAC) to access the Azure App Configuration service, so the connection string is not required.
+- `FOUNDATIONALLM_ENV`: This is required by the `OperationsManager` to disable the `verify` setting on requests. By setting the value to `dev`, it allows calls to the State API from LangChain and other Python-based APIs when running locally (`FOUNDATIONALLM_ENV` = `dev`) by disabling the check for a valid SSL cert on requests. This is only necessary when running the State API locally. Otherwise, the setting should be set to `prod`.
 
 | Name | Value | Description |
 | ---- | ----- | ----------- |
 | FOUNDATIONALLM_APP_CONFIGURATION_URI | REDACTED | Azure App Configuration URI |
+| FOUNDATIONALLM_ENV | `dev` or `prod` | Environment specification. Acceptable values are `dev` and `prod`. Defaults to `prod` if not set. |
 
 ## Running the solution locally
 

src/dotnet/State/Services/StateService.cs

@@ -53,6 +53,7 @@ public async Task<LongRunningOperation> CreateLongRunningOperation(string operat
             var operation = new LongRunningOperation
             {
                 Status = OperationStatus.Pending,
+                StatusMessage = "Operation was submitted and is pending execution.",
                 OperationId = operationId
             };
             return await cosmosDbService.UpsertLongRunningOperation(operation);

src/python/AgentHubAPI/app/dependencies.py

@@ -45,7 +45,7 @@ def validate_api_key_header(x_api_key: str = Depends(APIKeyHeader(name='X-API-Ke
         Returns True of the X-API-Key value from the request header matches the expected value.
         Otherwise, returns False.
     """
-    result = x_api_key == get_config().get_value(f'FoundationaLLM:APIs:{API_NAME}:APIKey')
+    result = x_api_key == get_config().get_value(f'FoundationaLLM:APIEndpoints:{API_NAME}:APIKey')
 
     if not result:
         logging.error('Invalid API key. You must provide a valid API key in the X-API-KEY header.')

src/python/DataSourceHubAPI/app/dependencies.py

@@ -46,7 +46,7 @@ def validate_api_key_header(x_api_key: str = Depends(APIKeyHeader(name='X-API-Ke
         Otherwise, returns False.
     """
 
-    result = x_api_key == get_config().get_value(f'FoundationaLLM:APIs:{API_NAME}:APIKey')
+    result = x_api_key == get_config().get_value(f'FoundationaLLM:APIEndpoints:{API_NAME}:APIKey')
 
     if not result:
         logging.error('Invalid API key. You must provide a valid API key in the X-API-KEY header.')

src/python/GatekeeperIntegrationAPI/app/dependencies.py

@@ -46,7 +46,7 @@ def validate_api_key_header(x_api_key: str = Depends(APIKeyHeader(name='X-API-Ke
         Otherwise, returns False.
     """
 
-    result = x_api_key == get_config().get_value(f'FoundationaLLM:APIs:{API_NAME}:APIKey')
+    result = x_api_key == get_config().get_value(f'FoundationaLLM:APIEndpoints:{API_NAME}:APIKey')
 
     if not result:
         logging.error('Invalid API key. You must provide a valid API key in the X-API-KEY header.')

src/python/LangChainAPI/app/dependencies.py

@@ -45,7 +45,7 @@ async def validate_api_key_header(x_api_key: str = Depends(APIKeyHeader(name='X-
         Otherwise, returns False.
     """
 
-    result = x_api_key == get_config().get_value(f'FoundationaLLM:APIs:{API_NAME}:APIKey')
+    result = x_api_key == get_config().get_value(f'FoundationaLLM:APIEndpoints:{API_NAME}:APIKey')
 
     if not result:
         logging.error('Invalid API key. You must provide a valid API key in the X-API-KEY header.')

src/python/LangChainAPI/app/routers/completions.py

@@ -14,7 +14,7 @@
     Response,
     status
 )
-from foundationallm.config import Configuration, Context
+from foundationallm.config import Configuration
 from foundationallm.models.operations import (
     LongRunningOperation,
     LongRunningOperationLogEntry,
@@ -130,14 +130,11 @@ async def create_completion_response(
             span.set_attribute('user_identity', x_user_identity)
 
             # Change the operation status to 'InProgress' using an async task.
-            loop = asyncio.get_running_loop()
-            loop.create_task(
-                operations_manager.update_operation(
-                    operation_id,
-                    instance_id,
-                    status = OperationStatus.INPROGRESS,
-                    status_message = f'Operation state changed to {OperationStatus.INPROGRESS}.'
-                )
+            await operations_manager.update_operation(
+                operation_id,
+                instance_id,
+                status = OperationStatus.INPROGRESS,
+                status_message = 'Operation state changed to in progress.'
             )
 
             # Create an orchestration manager to process the completion request.
@@ -162,15 +159,25 @@ async def create_completion_response(
                 )
             )            
         except Exception as e:
-            # TODO: Log the error and return an appropriate response to the caller.
             # Send the completion response to the State API and mark the operation as failed.
             print(f'Operation {operation_id} failed with error: {e}')
-            await operations_manager.update_operation(
-                operation_id,
-                instance_id,
-                status = OperationStatus.FAILED,
-                status_message = f'Operation failed with error: {e}'
+            completion = CompletionResponse(
+                operation_id = operation_id,
+                user_prompt=completion_request.user_prompt,
+                completion=f'Operation failed with error: {e}'
             )
+            await asyncio.gather(
+                operations_manager.set_operation_result(
+                    operation_id=operation_id,
+                    instance_id=instance_id,
+                    completion_response=completion),
+                operations_manager.update_operation(
+                    operation_id=operation_id,
+                    instance_id=instance_id,
+                    status = OperationStatus.FAILED,
+                    status_message = f'Operation failed with error: {e}'
+                )
+            )           
 
 @router.get(
     '/async-completions/{operation_id}/status',
@@ -192,8 +199,8 @@ async def get_operation_status(
         try:
             span.set_attribute('operation_id', operation_id)
             span.set_attribute('instance_id', instance_id)
-            
-            operation = operations_manager.get_operation_state(
+
+            operation = await operations_manager.get_operation(
                 operation_id,
                 instance_id
             )
@@ -226,7 +233,7 @@ async def get_operation_result(
             span.set_attribute('operation_id', operation_id)
             span.set_attribute('instance_id', instance_id)
 
-            completion_response = operations_manager.get_operation_result(
+            completion_response = await operations_manager.get_operation_result(
                 operation_id,
                 instance_id
             )
@@ -239,7 +246,7 @@ async def get_operation_result(
             handle_exception(e)
 
 @router.get(
-    '/async-completions/{operation_id}/log',
+    '/async-completions/{operation_id}/logs',
     summary = 'Retrieve the log of operational steps for the specified operation ID.',
     responses = {
         200: {'description': 'The operation log was retrieved successfully.'},
@@ -259,7 +266,7 @@ async def get_operation_log(
             span.set_attribute('operation_id', operation_id)
             span.set_attribute('instance_id', instance_id)
 
-            log = operations_manager.get_operation_log(
+            log = await operations_manager.get_operation_log(
                 operation_id,
                 instance_id
             )

src/python/PromptHubAPI/app/dependencies.py

@@ -46,7 +46,7 @@ def validate_api_key_header(x_api_key: str = Depends(APIKeyHeader(name='X-API-Ke
         Otherwise, returns False.
     """
 
-    result = x_api_key == get_config().get_value(f'FoundationaLLM:APIs:{API_NAME}:APIKey')
+    result = x_api_key == get_config().get_value(f'FoundationaLLM:APIEndpoints:{API_NAME}:APIKey')
 
     if not result:
         logging.error('Invalid API key. You must provide a valid API key in the X-API-KEY header.')

src/python/PythonSDK/foundationallm/langchain/agents/langchain_knowledge_management_agent.py

@@ -48,9 +48,9 @@ def get_prediction_from_audio_file(self, attachments: Optional[List[str]] = None
 
         try:
             storage_manager = BlobStorageManager(
-                account_name=self.config.get_value('FoundationaLLM:Attachment:ResourceProviderService:Storage:AccountName'),
+                account_name=self.config.get_value('FoundationaLLM:ResourceProviders:Attachment:Storage:AccountName'),
                 container_name=file.split('/')[0],
-                authentication_type=self.config.get_value('FoundationaLLM:Attachment:ResourceProviderService:Storage:AuthenticationType')
+                authentication_type=self.config.get_value('FoundationaLLM:ResourceProviders:Attachment:Storage:AuthenticationType')
             )
         except Exception as e:
             raise e
@@ -79,8 +79,8 @@ def get_prediction_from_audio_file(self, attachments: Optional[List[str]] = None
         audio_embeddings = clap_model.get_audio_embeddings(file_paths, resample=True)
         data = audio_embeddings.numpy().tolist()
 
-        base_url = self.config.get_value('FoundationaLLM:APIs:AudioClassificationAPI:APIUrl').rstrip('/')
-        endpoint = self.config.get_value('FoundationaLLM:APIs:AudioClassificationAPI:Classification:PredictionEndpoint')
+        base_url = self.config.get_value('FoundationaLLM:APIEndpoints:AudioClassificationAPI:APIUrl').rstrip('/')
+        endpoint = self.config.get_value('FoundationaLLM:APIEndpoints:AudioClassificationAPI:Classification:PredictionEndpoint')
         api_endpoint = f'{base_url}{endpoint}'
 
         # Create the embeddings payload.

src/python/PythonSDK/foundationallm/models/operations/long_running_operation.py

@@ -1,12 +1,13 @@
 from datetime import datetime
 from pydantic import BaseModel, Field
+from typing import Optional
 
 class LongRunningOperation(BaseModel):
     """
     Class representing a long running operation.
     """
     operation_id: str = Field(description='The unique identifier for the operation.')
     status: str = Field(description='The status of the operation.')
-    status_message: str = Field(description='The message associated with the operation status.')
-    last_updated: datetime = Field(default=datetime.now(), description='The timestamp of the last update to the operation.')
+    status_message: Optional[str] = Field(description='The message associated with the operation status.')
+    last_updated: Optional[datetime] = Field(default=None, description='The timestamp of the last update to the operation.')
 

src/python/PythonSDK/foundationallm/models/operations/long_running_operation_log_entry.py

@@ -4,6 +4,9 @@ class LongRunningOperationLogEntry(BaseModel):
     """
     Represents an entry in the operation execution log.
     """
-    timestamp: str = Field(description='The timestamp of the log entry.')
+    id: str = Field(description='The unique id of the log entry.')
+    type: str = Field(description='The type of log entry')
+    operation_id: str = Field(description='The unique id of the operation.')
+    time_stamp: str = Field(description='The timestamp of the log entry.')
     status: str = Field(description='The status of the operation at the time of the log entry.')
     status_message: str = Field(description='The status message of the log entry.')