Fleshed out LLM Proxy

examples/portfolio/agents/advisor_agent.py

@@ -2,23 +2,19 @@
 #
 # Final stage. Turns the computed portfolio metrics and risk figures into a
 # short, plain-English briefing using a small, cheap model on AWS Bedrock
-# (Converse API), called via ventis.llm.bedrock so token/cost telemetry gets
-# recorded onto this execution's future:<future_id> hash. Configure
-# with env vars:
+# (Converse API). Token/cost telemetry gets recorded automatically via the
+# LLM proxy. Configure with env vars:
 #   BEDROCK_MODEL_ID  (default: meta.llama3-8b-instruct-v1:0)
 #   AWS_REGION        (default: us-east-1)
+#   AWS_ENDPOINT_URL_BEDROCK_RUNTIME (routes to proxy for telemetry)
 #
 # If the LLM is unavailable (returns an empty string), it falls back to a
 # deterministic templated summary so the pipeline still returns.
 #
 # Resource profile: cheap CPU; the LLM cost sits in the Bedrock call, not here.
 
 import os
-
-try:
-    from ventis.llm.bedrock import call_bedrock
-except ImportError:
-    from bedrock import call_bedrock
+import boto3
 
 
 class AdvisorAgent(object):
@@ -28,16 +24,16 @@ def __init__(self):
             "BEDROCK_MODEL_ID", "meta.llama3-8b-instruct-v1:0"
         )
         self.region = os.environ.get("AWS_REGION", "us-east-1")
+        self._client = boto3.client("bedrock-runtime", region_name=self.region)
 
     def summarize(self, holdings: dict, metrics: dict, risk: dict) -> str:
         """Write a short plain-English briefing on the portfolio."""
         prompt = self._build_prompt(holdings, metrics, risk)
         try:
-            response = call_bedrock(
-                model_id=self.model_id,
+            response = self._client.converse(
+                modelId=self.model_id,
                 messages=[{"role": "user", "content": [{"text": prompt}]}],
-                inference_config={"maxTokens": 400, "temperature": 0.2},
-                region=self.region,
+                inferenceConfig={"maxTokens": 400, "temperature": 0.2},
             )
             return response["output"]["message"]["content"][0]["text"]
         except Exception as e:

examples/portfolio/agents/intent_agent.py

@@ -7,9 +7,8 @@
 #      ->  {"holdings": {"AAPL": 0.4, "MSFT": 0.35, "NVDA": 0.25},
 #           "lookback_days": 180}
 #
-# Calls AWS Bedrock (Converse API) via ventis.llm.bedrock -- same pattern as
-# AdvisorAgent -- so token/cost telemetry gets recorded onto this execution's
-# future:<future_id> hash. Configure with env vars:
+# Calls AWS Bedrock (Converse API) via standard boto3. Telemetry is collected
+# automatically by the LLM proxy. Configure with env vars:
 #   BEDROCK_MODEL_ID  (default: meta.llama3-8b-instruct-v1:0)
 #   AWS_REGION        (default: us-east-1)
 #
@@ -23,11 +22,7 @@
 import os
 import re
 import json
-
-try:
-    from ventis.llm.bedrock import call_bedrock
-except ImportError:
-    from bedrock import call_bedrock
+import boto3
 
 DEFAULT_LOOKBACK_DAYS = 365
 
@@ -39,14 +34,14 @@ def __init__(self):
             "BEDROCK_MODEL_ID", "meta.llama3-8b-instruct-v1:0"
         )
         self.region = os.environ.get("AWS_REGION", "us-east-1")
+        self._client = boto3.client("bedrock-runtime", region_name=self.region)
 
     def parse(self, query: str) -> dict:
         """Parse a natural-language portfolio request into holdings + lookback."""
-        response = call_bedrock(
-            model_id=self.model_id,
+        response = self._client.converse(
+            modelId=self.model_id,
             messages=[{"role": "user", "content": [{"text": self._build_prompt(query)}]}],
-            inference_config={"maxTokens": 300, "temperature": 0.0},
-            region=self.region,
+            inferenceConfig={"maxTokens": 300, "temperature": 0.0},
         )
         text = response["output"]["message"]["content"][0]["text"]
         if not text:

examples/portfolio/config/global_controller.yaml

@@ -78,6 +78,20 @@ agents:
     provider: EC2
     instance_type: t3.micro
 
+
+otel:
+  destinations:
+    - name: railway
+      protocol: grpc
+      endpoint: ${RAILWAY_OTLP_ENDPOINT}
+      insecure: true
+      headers: {}
+    - name: grafana
+      protocol: http
+      endpoint: ${GRAFANA_OTLP_ENDPOINT}/v1/traces
+      headers:
+        Authorization: Basic ${GRAFANA_OTLP_HEADERS}
+
 # Polling interval in seconds
 poll_interval: 5
 
@@ -86,3 +100,16 @@ redis:
   host: localhost
   port: 6379
   db: 0
+
+# EC2 defaults for `provider: EC2` replicas.
+ec2:
+  region: ${EC2_REGION}
+  ami_id: ${EC2_AMI_ID}
+  subnet_id: ${EC2_SUBNET_ID}
+  security_group_ids:
+    - ${EC2_SECURITY_GROUP_ID}
+  ssh_user: ${EC2_SSH_USER}
+  ssh_private_key_path: ${EC2_SSH_PRIVATE_KEY_PATH}
+
+database:
+  url: ${DATABASE_URL}

examples/text2sql/agents/vllm_agent.py

@@ -1,38 +1,32 @@
 # VLLM Agent
 #
 # LLM backend for SQL candidate generation, called remotely by
-# SQLGeneratorAgent. Calls AWS Bedrock (Converse API) via ventis.llm.bedrock
-# so token/cost telemetry gets recorded onto this execution's
-# future:<future_id> hash — same pattern as
-# examples/portfolio/agents/advisor_agent.py.
+# SQLGeneratorAgent. Calls AWS Bedrock (Converse API) via standard boto3.
+# Telemetry is collected automatically by the LLM proxy.
 # Configure with env vars:
 #   BEDROCK_MODEL_ID  (default: meta.llama3-8b-instruct-v1:0)
 #   AWS_REGION        (default: us-east-1)
 #
 # Falls back to a synthetic placeholder response if Bedrock is unavailable.
 
 import os
-
-try:
-    from ventis.llm.bedrock import call_bedrock
-except ImportError:
-    from bedrock import call_bedrock
+import boto3
 
 
 class VllmAgent(object):
     def __init__(self):
         self.tools = [self.generate]
         self.model_id = os.environ.get("BEDROCK_MODEL_ID", "meta.llama3-8b-instruct-v1:0")
         self.region = os.environ.get("AWS_REGION", "us-east-1")
+        self._client = boto3.client("bedrock-runtime", region_name=self.region)
 
     def generate(self, prompt: str) -> str:
         """Generates a response using an LLM model based on the given prompt."""
         try:
-            response = call_bedrock(
-                model_id=self.model_id,
+            response = self._client.converse(
+                modelId=self.model_id,
                 messages=[{"role": "user", "content": [{"text": prompt}]}],
-                inference_config={"maxTokens": 400, "temperature": 0.2},
-                region=self.region,
+                inferenceConfig={"maxTokens": 400, "temperature": 0.2},
             )
             return response["output"]["message"]["content"][0]["text"]
         except Exception as e:

ventis/controller/cloud_provider_logic/EC2/_runtime.py

@@ -277,6 +277,8 @@ def _bootstrap_instance(host, spec, replica_index, cfg, redis_host, redis_port,
         f"VENTIS_AGENT_PORT={CONTAINER_PORT}",
         "-e",
         f"VENTIS_POLL_INTERVAL={_controller.config.get('poll_interval', 5)}",
+        "-e",
+        "AWS_ENDPOINT_URL_BEDROCK_RUNTIME=http://127.0.0.1:8081/bedrock",
     ]
     if spec.get("type") == "workflow":
         db_url = _controller.config.get("database", {}).get("url")

ventis/controller/cloud_provider_logic/Local/_runtime.py

@@ -94,6 +94,8 @@ def bootstrap_instance(provisioned, spec, replica_index, agent_id):
         f"VENTIS_REDIS_PORT={spec.get('redis_port', 6379)}",
         "-e",
         f"VENTIS_POLL_INTERVAL={_require_controller().config.get('poll_interval', 5)}",
+        "-e",
+        "AWS_ENDPOINT_URL_BEDROCK_RUNTIME=http://127.0.0.1:8081/bedrock",
     ]
     if ctrl_type == "workflow":
         cmd.extend(["-p", f"{spec.get('api_port', 8080)}:8080"])

ventis/controller/local_controller.py

@@ -6,6 +6,7 @@
 import logging
 import os
 import random
+import subprocess
 import sys
 import threading
 import time
@@ -35,6 +36,16 @@
     import ventis.ventis_context as ventis_context
 except ImportError:
     import ventis_context
+
+# Auto-inject Ventis headers into all boto3 Bedrock calls
+try:
+    from ventis.llm_proxy import proxy
+except ImportError:
+    try:
+        from llm_proxy import proxy
+    except ImportError:
+        pass  # No proxy available, agents will call Bedrock directly
+
 import local_controler_pb2
 import local_controler_pb2_grpc
 
@@ -102,6 +113,9 @@ def __init__(self, port=50051):
         max_instances = int(os.environ.get("VENTIS_MAX_AGENT_INSTANCES", 8))
         self._executor = ThreadPoolExecutor(max_workers=max_instances)
 
+        # Start LLM proxy in this container
+        self._proxy_process = self._start_llm_proxy(redis_host, redis_port)
+
         logger.info(
             "Local controller initialized at %s (max_agent_instances=%d), reported healthy to Redis.",
             self._my_endpoint,
@@ -111,6 +125,31 @@ def __init__(self, port=50051):
         # Load the agent class dynamically
         self.agent = self._load_agent()
 
+    def _start_llm_proxy(self, redis_host, redis_port):
+        """Start LLM proxy as a subprocess in this container."""
+        try:
+            import subprocess
+            proxy_env = os.environ.copy()
+            proxy_env.update({
+                "PROXY_HOST": "127.0.0.1",
+                "PROXY_PORT": "8081",
+                "VENTIS_REDIS_HOST": redis_host,
+                "VENTIS_REDIS_PORT": str(redis_port),
+            })
+
+            # Start proxy as background subprocess
+            proxy_process = subprocess.Popen(
+                [sys.executable, "-m", "ventis.llm_proxy"],
+                env=proxy_env,
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.DEVNULL,
+            )
+            logger.info("Started LLM proxy on 127.0.0.1:8081 (PID: %d)", proxy_process.pid)
+            return proxy_process
+        except Exception as e:
+            logger.warning("Failed to start LLM proxy: %s", e)
+            return None
+
     def _collect_metrics(self):
         """Snapshot current instance health/resource metrics.