[Integration] Integrating LLM-proxy (#54/#63) onto PR stack

README.md

@@ -76,7 +76,7 @@ Edit `.car/config/global_controller.yaml` in your project directory to list the
 Agents that need API keys read them from environment variables. Point `env_file` at a `.env` file to have Ventis inject it into every agent container:
 
 ```yaml
-# config/global_controller.yaml
+# .car/config/global_controller.yaml
 env_file: .env
 ```
 

cli/README.md

@@ -23,4 +23,4 @@ leaving every other line unchanged.
 
 If you have a workflow running, and want to make a config change, canyonos config automatically would reload the project with your config. If you change the workflow files itself though and want the changes to take effect, you need to redeploy from scratch, running canyonos build for good measure
 
-# Use: canyonos -h
+# Use: canyonos -h

examples/portfolio/agents/advisor_agent.py

@@ -2,8 +2,9 @@
 #
 # 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.controller.bedrock so token/cost telemetry gets
-# recorded onto this execution's future:<future_id> hash. Configure
+# (Converse API), called directly via boto3. Token/cost telemetry is recorded
+# onto this execution's future:<future_id> hash transparently by the Ventis LLM
+# proxy each agent container's boto3 calls are routed through. Configure
 # with env vars:
 #   BEDROCK_MODEL_ID  (default: meta.llama3-8b-instruct-v1:0)
 #   AWS_REGION        (default: us-east-1)
@@ -15,10 +16,7 @@
 
 import os
 
-try:
-    from ventis.controller.bedrock import call_bedrock
-except ImportError:
-    from bedrock import call_bedrock
+import boto3
 
 
 class AdvisorAgent(object):
@@ -28,16 +26,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,11 @@
 #      ->  {"holdings": {"AAPL": 0.4, "MSFT": 0.35, "NVDA": 0.25},
 #           "lookback_days": 180}
 #
-# Calls AWS Bedrock (Converse API) via ventis.controller.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) directly via boto3 -- same pattern as
+# AdvisorAgent. Token/cost telemetry is recorded onto this execution's
+# future:<future_id> hash transparently by the Ventis LLM proxy, which each
+# agent container's boto3 calls are routed through (AWS_ENDPOINT_URL_BEDROCK_RUNTIME).
+# Configure with env vars:
 #   BEDROCK_MODEL_ID  (default: meta.llama3-8b-instruct-v1:0)
 #   AWS_REGION        (default: us-east-1)
 #
@@ -24,10 +26,7 @@
 import re
 import json
 
-try:
-    from ventis.controller.bedrock import call_bedrock
-except ImportError:
-    from bedrock import call_bedrock
+import boto3
 
 DEFAULT_LOOKBACK_DAYS = 365
 
@@ -39,14 +38,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

@@ -7,7 +7,7 @@
 
 agents:
   # Stage 0: parse the free-text request into structured holdings + lookback
-  # window (calls Bedrock directly via ventis.llm.bedrock). Cheap CPU, one
+  # window (calls Bedrock via boto3, routed through the Ventis LLM proxy). Cheap CPU, one
   # call per request, on the critical path before the fan-out.
   - name: IntentAgent
     redis_port: 6379

examples/text2sql/agents/vllm_agent.py

@@ -1,9 +1,10 @@
 # VLLM Agent
 #
 # LLM backend for SQL candidate generation, called remotely by
-# SQLGeneratorAgent. Calls AWS Bedrock (Converse API) via ventis.controller.bedrock
-# so token/cost telemetry gets recorded onto this execution's
-# future:<future_id> hash — same pattern as
+# SQLGeneratorAgent. Calls AWS Bedrock (Converse API) directly via boto3.
+# Token/cost telemetry is recorded onto this execution's future:<future_id>
+# hash transparently by the Ventis LLM proxy each agent container's boto3 calls
+# are routed through — same pattern as
 # examples/portfolio/agents/advisor_agent.py.
 # Configure with env vars:
 #   BEDROCK_MODEL_ID  (default: meta.llama3-8b-instruct-v1:0)
@@ -13,26 +14,23 @@
 
 import os
 
-try:
-    from ventis.controller.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:

pyproject.toml

@@ -12,6 +12,7 @@ dependencies = [
     "psycopg[binary]",
     "pyyaml",
     "flask",
+    "requests",
     "psutil",
     "opentelemetry-api>=1.44.0",
     "opentelemetry-sdk>=1.44.0",

requirements.txt

@@ -4,6 +4,7 @@ grpcio-tools
 redis
 pyyaml
 flask
+requests
 sqlalchemy
 psycopg[binary]
 psutil

tests/test_instance_manager_runtime.py

@@ -170,6 +170,8 @@ def test_local_instances_keep_default_host_and_increment_host_ports(self):
                     "VENTIS_REDIS_PORT=6379",
                     "-e",
                     "VENTIS_POLL_INTERVAL=5",
+                    "-e",
+                    "AWS_ENDPOINT_URL_BEDROCK_RUNTIME=http://127.0.0.1:8081/bedrock",
                     "ventis-alpha",
                 ],
                 "localhost",
@@ -226,6 +228,8 @@ def test_local_workflow_and_resource_flags_stay_the_same(self):
                     "VENTIS_REDIS_PORT=6379",
                     "-e",
                     "VENTIS_POLL_INTERVAL=5",
+                    "-e",
+                    "AWS_ENDPOINT_URL_BEDROCK_RUNTIME=http://127.0.0.1:8081/bedrock",
                     "-p",
                     "8080:8080",
                     "--cpus",

ventis/FUTURE_SCHEMA.md

@@ -19,16 +19,16 @@ Fields currently written into `future:{future_id}`, and where:
 | `created_at`                | `future.py` only (origin submission time) |
 | `result`                    | `future.py`, `local_controller.py` |
 | `failed`                    | `future.py`, `local_controller.py` |
-| `error`                     | `future.py` (`_submit_request`), `local_controller.py` (`_mark_future_failed`) -- the sole failure-message field; `bedrock.py` deliberately never writes it |
+| `error`                     | `future.py` (`_submit_request`), `local_controller.py` (`_mark_future_failed`) -- the sole failure-message field; the LLM proxy deliberately never writes it |
 | `finished_at`               | `local_controller.py` (`_execute_locally` finally block) |
 | `cpu_resource`              | `local_controller.py` |
 | `gpu_resource`              | `local_controller.py` |
 | `agent`                     | `local_controller.py` (agent_id that executed this step) |
 | `queue_time`                | `local_controller.py` (only when `submitted_at` is known) |
-| `model`                     | `llm/bedrock.py` (`call_bedrock`) |
-| `input_token_count`         | `llm/bedrock.py` |
-| `output_token_count`        | `llm/bedrock.py` |
-| `token_count`               | `llm/bedrock.py` |
-| `errors`                    | `llm/bedrock.py` (Bedrock call error count) |
-| `input_cache_tokens`        | `llm/bedrock.py` |
-| `input_cache_write_tokens`  | `llm/bedrock.py` |
+| `model`                     | `llm_proxy/hooks.py` (on_response) |
+| `input_token_count`         | `llm_proxy/hooks.py` |
+| `output_token_count`        | `llm_proxy/hooks.py` |
+| `token_count`               | `llm_proxy/hooks.py` |
+| `errors`                    | `llm_proxy/hooks.py` (Bedrock call error flag) |
+| `input_cache_tokens`        | `llm_proxy/hooks.py` |
+| `input_cache_write_tokens`  | `llm_proxy/hooks.py` |

ventis/controller/cloud_provider_logic/EC2/_runtime.py

@@ -288,6 +288,10 @@ 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)}",
+        # Route the agent's boto3 Bedrock calls through the in-container LLM
+        # proxy (started by LocalController) so token/cost telemetry is captured.
+        "-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

@@ -93,6 +93,10 @@ 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)}",
+            # Route the agent's boto3 Bedrock calls through the in-container LLM
+            # proxy (started by LocalController) so token/cost telemetry is captured.
+            "-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

@@ -35,6 +35,18 @@
     import ventis.controller.ventis_context as ventis_context
 except ImportError:
     import ventis_context
+
+# Auto-inject X-Ventis-Future-ID into all boto3 Bedrock calls so the LLM proxy
+# can attribute token/cost telemetry to the executing future. Import for its
+# global boto3 event-hook side effect; safe no-op if the proxy isn't present.
+try:
+    from ventis.llm_proxy import proxy as _llm_proxy_autoinject  # noqa: F401
+except ImportError:
+    try:
+        from llm_proxy import proxy as _llm_proxy_autoinject  # noqa: F401
+    except ImportError:
+        pass  # No proxy available; agents call Bedrock directly.
+
 import local_controler_pb2
 import local_controler_pb2_grpc
 
@@ -102,6 +114,11 @@ def __init__(self, port=50051):
         max_instances = int(os.environ.get("VENTIS_MAX_AGENT_INSTANCES", 8))
         self._executor = ThreadPoolExecutor(max_workers=max_instances)
 
+        # Start the LLM proxy alongside the agent in this container. Bedrock
+        # calls are routed to it via AWS_ENDPOINT_URL_BEDROCK_RUNTIME (injected
+        # by the runtime), and it writes token/cost telemetry to Redis.
+        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 +128,33 @@ 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 the LLM proxy as a subprocess in this container (127.0.0.1:8081).
+
+        Best-effort: a failure here must never stop the controller from coming up.
+        """
+        import subprocess
+
+        try:
+            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),
+            })
+            proxy_process = subprocess.Popen(
+                [sys.executable, "-m", "ventis.llm_proxy"],
+                env=proxy_env,
+            )
+            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.