Stabilize the connector and add monitoring

examples/llm/anthropic_chat.yaml

@@ -12,7 +12,7 @@
 # It will then send an event back to Solace with the topic: `demo/question/response`
 #
 # Dependencies:
-# pip install -U langchain-anthropic
+# pip install -U langchain-anthropic langchain-core~=0.3.0 langchain~=0.3.0
 #
 # required ENV variables:
 # - ANTHROPIC_API_KEY

examples/llm/bedrock_anthropic_chat.yaml

@@ -11,7 +11,7 @@
 # }
 # 
 # Dependencies:
-# pip install langchain_aws
+# pip install langchain_aws langchain-core~=0.3.0 langchain~=0.3.0
 # 
 # required ENV variables:
 # - AWS_BEDROCK_ANTHROPIC_CLAUDE_MODEL_ID

examples/llm/langchain_openai_with_history_chat.yaml

@@ -12,7 +12,7 @@
 # It will then send an event back to Solace with the topic: `demo/joke/subject/response`
 #
 # Dependencies:
-# pip install -U langchain_openai openai
+# pip install -U langchain_openai openai langchain-core~=0.3.0 langchain~=0.3.0
 #
 # required ENV variables:
 # - OPENAI_API_KEY

examples/llm/litellm_chat.yaml

@@ -33,9 +33,16 @@
 
 ---
 log:
-  stdout_log_level: INFO
+  stdout_log_level: DEBUG
   log_file_level: DEBUG
-  log_file: solace_ai_connector.log
+  log_file: ${LOG_FILE}
+  log_format: jsonl
+  logback: 
+    rollingpolicy:
+      file-name-pattern: "${LOG_FILE}.%d{yyyy-MM-dd}.%i.gz"
+      max-file-size: 100MB
+      max-history: 5
+      total-size-cap: 1GB
 
 shared_config:
   - broker_config: &broker_connection
@@ -44,6 +51,8 @@ shared_config:
       broker_username: ${SOLACE_BROKER_USERNAME}
       broker_password: ${SOLACE_BROKER_PASSWORD}
       broker_vpn: ${SOLACE_BROKER_VPN}
+      reconnection_strategy: forever_retry # options: forever_retry, parametrized_retry
+      retry_interval: 10000 # in milliseconds
 
 # Take from input broker and publish back to Solace
 flows:

examples/llm/mixture_of_agents.yaml

@@ -11,7 +11,7 @@
 # NOTE: For horizontal scaling, partitioned queues must be used. This is not implemented in this example.
 #
 # Dependencies:
-# pip install -U langchain-google-vertexai langchain_anthropic langchain_openai openai
+# pip install -U langchain-google-vertexai langchain_anthropic langchain_openai openai langchain-core~=0.3.0 langchain~=0.3.0
 #
 # required ENV variables:
 # - GOOGLE_APPLICATION_CREDENTIALS: the path to a service account JSON file

examples/llm/openai_chat.yaml

@@ -11,7 +11,7 @@
 # It will then send an event back to Solace with the topic: `demo/question/response`
 #
 # Dependencies:
-# pip install -U langchain_openai openai
+# pip install -U langchain_openai openai langchain-core~=0.3.0 langchain~=0.3.0
 #
 # required ENV variables:
 # - OPENAI_API_KEY

examples/llm/openai_chroma_rag.yaml

@@ -7,7 +7,7 @@
 # Load Data:
 # Send data to Solace topic `demo/rag/data` with the following payload format:
 # {
-#   "texts": [<text data 1>. <text data 2>, ...]
+#   "text": text
 # }
 #
 # RAG Query:
@@ -18,7 +18,7 @@
 # The response will be sent to Solace topic `demo/rag/query/response`
 #
 # Dependencies:
-# pip install -U langchain_openai openai chromadb langchain-chroma
+# pip install -U langchain_openai openai chromadb langchain-chroma langchain-core~=0.3.0 langchain~=0.3.0
 #
 # Required ENV variables:
 # - OPENAI_API_KEY
@@ -61,6 +61,22 @@ flows:
           payload_encoding: utf-8
           payload_format: json
 
+      # Split text
+      - component_name: text_splitter
+        component_module: langchain_split_text
+        component_config:
+          langchain_module: langchain_text_splitters
+          langchain_class: TokenTextSplitter
+          langchain_component_config:
+            chunk_size: 10
+            chunk_overlap: 1
+        input_transforms:
+          - type: copy
+            source_expression: input.payload:text
+            dest_expression: user_data.input:text
+        input_selection:
+          source_expression: user_data.input
+
       # Embedding data & ChromaDB ingest
       - component_name: chroma_embed
         component_module: langchain_vector_store_embedding_index
@@ -81,11 +97,29 @@ flows:
             source_value: topic:demo/rag/data
             dest_expression: user_data.vector_input:metadatas.source
           - type: copy
-            source_expression: input.payload:texts
+            source_expression: previous
             dest_expression: user_data.vector_input:texts
         input_selection:
           source_expression: user_data.vector_input
 
+            # Send response back to broker
+      - component_name: send_response
+        component_module: broker_output
+        component_config:
+          <<: *broker_connection
+          payload_encoding: utf-8
+          payload_format: json
+          copy_user_properties: true
+        input_transforms:
+          - type: copy
+            source_expression: previous
+            dest_expression: user_data.output:payload
+          - type: copy
+            source_expression: template:demo/rag/response
+            dest_expression: user_data.output:topic
+        input_selection:
+          source_expression: user_data.output
+
   # RAG Inference flow
   - name: OpenAI_RAG
     components:

examples/llm/openai_component_request_response.yaml

@@ -28,7 +28,7 @@
 # It will then send an event back to Solace with the topic: `demo/question/response`
 #
 # Dependencies:
-# pip install -U langchain_openai openai
+# pip install -U langchain_openai openai langchain-core~=0.3.0 langchain~=0.3.0
 #
 # required ENV variables:
 # - OPENAI_API_KEY

examples/llm/vertexai_chat.yaml

@@ -11,7 +11,7 @@
 # It will then send an event back to Solace with the topic: `demo/question/response`
 #
 # Dependencies:
-# pip install -U langchain-google-vertexai
+# pip install -U langchain-google-vertexai langchain-core~=0.3.0 langchain~=0.3.0
 #
 # required ENV variables:
 # - GOOGLE_APPLICATION_CREDENTIALS: the path to a service account JSON file

pyproject.toml

@@ -19,12 +19,9 @@ classifiers = [
 ]
 dependencies = [
     "boto3~=1.34.122",
-    "langchain-core~=0.3.0",
-    "langchain~=0.3.0",
     "PyYAML~=6.0.1",
     "Requests~=2.32.3",
-    "solace_pubsubplus>=1.8.0",
-    "litellm>=1.51.3",
+    "solace_pubsubplus>=1.9.0",
     "Flask~=3.0.3",
     "Flask-SocketIO~=5.4.1",
     "build~=1.2.2.post1",
@@ -44,7 +41,7 @@ solace-ai-connector-gen-docs = "solace_ai_connector.tools.gen_component_docs:mai
 [tool.hatch.envs.hatch-test]
 installer = "pip"
 
-# # Specify minimum and maximum Python versions to test
+# Specify minimum and maximum Python versions to test
 [[tool.hatch.envs.hatch-test.matrix]]
 python = ["3.10", "3.12"]
 

requirements.txt

@@ -1,10 +1,9 @@
 boto3~=1.34.122
-langchain-core~=0.3.0
-langchain~=0.3.0
 PyYAML~=6.0.1
 Requests~=2.32.3
-solace_pubsubplus~=1.8.0
-litellm~=1.51.3
+solace_pubsubplus~=1.9.0
 Flask~=3.0.3
 Flask-SocketIO~=5.4.1
-build~=1.2.2.post1
+build~=1.2.2.post1
+datadog~=0.50.2
+SQLAlchemy~=2.0.36

src/solace_ai_connector/common/log.py

@@ -2,6 +2,8 @@
 import logging
 import logging.handlers
 import json
+import os
+from datetime import datetime
 
 
 log = logging.getLogger("solace_ai_connector")
@@ -35,7 +37,22 @@ def format(self, record):
         return json.dumps(log_record)
 
 
-def setup_log(logFilePath, stdOutLogLevel, fileLogLevel, logFormat):
+def convert_to_bytes(size_str):
+    size_str = size_str.upper()
+    size_units = {"KB": 1024, "MB": 1024**2, "GB": 1024**3, "TB": 1024**4, "B": 1}
+    for unit in size_units:
+        if size_str.endswith(unit):
+            return int(size_str[: -len(unit)]) * size_units[unit]
+    return int(size_str)
+
+
+def setup_log(
+    logFilePath,
+    stdOutLogLevel,
+    fileLogLevel,
+    logFormat,
+    logBack,
+):
     """
     Set up the configuration for the logger.
 
@@ -44,8 +61,9 @@ def setup_log(logFilePath, stdOutLogLevel, fileLogLevel, logFormat):
         stdOutLogLevel (int): Logging level for standard output.
         fileLogLevel (int): Logging level for the log file.
         logFormat (str): Format of the log output ('jsonl' or 'pipe-delimited').
-
+        logBack (dict): Rolling log file configuration.
     """
+
     # Set the global logger level to the lowest of the two levels
     log.setLevel(min(stdOutLogLevel, fileLogLevel))
 
@@ -54,17 +72,61 @@ def setup_log(logFilePath, stdOutLogLevel, fileLogLevel, logFormat):
     stream_formatter = logging.Formatter("%(message)s")
     stream_handler.setFormatter(stream_formatter)
 
-    # Create an empty file at logFilePath (this will overwrite any existing content)
-    with open(logFilePath, "w") as file:
-        file.write("")
-
-    # file_handler = logging.handlers.TimedRotatingFileHandler(
-    #    filename=logFilePath, when='midnight', backupCount=30, mode='w')
-    file_handler = logging.FileHandler(filename=logFilePath, mode="a")
     if logFormat == "jsonl":
         file_formatter = JsonlFormatter()
     else:
         file_formatter = logging.Formatter("%(asctime)s |  %(levelname)s: %(message)s")
+
+    if logBack:
+        rollingpolicy = logBack.get("rollingpolicy", {})
+        if rollingpolicy:
+            if "file-name-pattern" not in rollingpolicy:
+                log.warning(
+                    "file-name-pattern is required in rollingpolicy. Continuing with default value '{LOG_FILE}.%d{yyyy-MM-dd}.%i'."
+                )
+            file_name_pattern = rollingpolicy.get(
+                "file-name-pattern", "{LOG_FILE}.%d{yyyy-MM-dd}.%i.gz"
+            )
+
+            if "max-file-size" not in rollingpolicy:
+                log.warning(
+                    "max-file-size is required in rollingpolicy. Continuing with default value '1GB'."
+                )
+            max_file_size = rollingpolicy.get("max-file-size", "1GB")
+
+            if "max-history" not in rollingpolicy:
+                log.warning(
+                    "max-history is required in rollingpolicy. Continuing with default value '7'."
+                )
+            max_history = rollingpolicy.get("max-history", 7)
+
+            if "total-size-cap" not in rollingpolicy:
+                log.warning(
+                    "total-size-cap is required in rollingpolicy. Continuing with default value '1TB'."
+                )
+            total_size_cap = rollingpolicy.get("total-size-cap", "1TB")
+
+            # Convert size strings to bytes
+            max_file_size = convert_to_bytes(max_file_size)
+            total_size_cap = convert_to_bytes(total_size_cap)
+
+            # Generate the log file name using the pattern
+            log_file_name = logFilePath
+
+            # Overwrite the file handler with a rotating file handler
+            file_handler = logging.handlers.RotatingFileHandler(
+                filename=log_file_name,
+                backupCount=max_history,
+                maxBytes=max_file_size,
+            )
+            file_handler.namer = (
+                lambda name: file_name_pattern.replace("${LOG_FILE}", logFilePath)
+                .replace("%d{yyyy-MM-dd}", datetime.now().strftime("%Y-%m-%d"))
+                .replace("%i", str(name.split(".")[-1]))
+            )
+    else:
+        file_handler = logging.FileHandler(filename=logFilePath, mode="a")
+
     file_handler.setFormatter(file_formatter)
     file_handler.setLevel(fileLogLevel)
 

src/solace_ai_connector/common/messaging/messaging_builder.py

@@ -6,18 +6,23 @@
 
 # Make a Messaging Service builder - this is a factory for Messaging Service objects
 class MessagingServiceBuilder:
-    def __init__(self, flow_lock_manager, flow_kv_store):
+
+    def __init__(self, flow_lock_manager, flow_kv_store, broker_name, stop_signal):
         self.broker_properties = {}
         self.flow_lock_manager = flow_lock_manager
         self.flow_kv_store = flow_kv_store
+        self.stop_signal = stop_signal
+        self.broker_name = broker_name
 
     def from_properties(self, broker_properties: dict):
         self.broker_properties = broker_properties
         return self
 
     def build(self):
         if self.broker_properties["broker_type"] == "solace":
-            return SolaceMessaging(self.broker_properties)
+            return SolaceMessaging(
+                self.broker_properties, self.broker_name, self.stop_signal
+            )
         elif self.broker_properties["broker_type"] == "dev_broker":
             return DevBroker(
                 self.broker_properties, self.flow_lock_manager, self.flow_kv_store