Telemetry signals

.dockerignore

@@ -19,3 +19,4 @@ AWSCLIV2.pkg
 docker_container
 grpc_stubs
 stubs
+tests

.gitignore

@@ -40,3 +40,4 @@ docker_container/
 !.env.example
 AWSCLIV2.pkg
 .python-version
+

README.md

@@ -26,6 +26,7 @@ Note: Installation of ventis only needs to be done on the machine where you are
 
 - **Python 3.10+**
 - **Docker** — Used to manage agents.
+- **Docker Buildx** (optional) — If available, `ventis build` builds all agent/workflow images in a single parallel `docker buildx bake` pass; otherwise it falls back to building them sequentially.
 
 ---
 

examples/config/global_controller.yaml → examples/finance/config/global_controller.yaml

@@ -34,7 +34,7 @@ agents:
     replicas: 1
     type: workflow
     redis_port: 6379
-    workflow_file: workflows/example_workflow.py
+    workflow_file: workflow/example_workflow.py
     provider: local
 
 poll_interval: 5

ventis/templates/README.md → examples/helloworld/README.md

@@ -26,7 +26,7 @@ curl http://<workflow_host_ip>:8080/status/<request_id>
 ├── agents/               # Agent implementations and YAML definitions
 │   ├── example_agent.py
 │   └── example_agent.yaml
-├── workflows/            # Workflow scripts (deployed as REST APIs)
+├── workflow/             # Workflow scripts (deployed as REST APIs)
 │   └── example_workflow.py
 ├── config/
 │   ├── global_controller.yaml   # Deployment configuration

ventis/templates/config/global_controller.yaml → examples/helloworld/config/global_controller.yaml

@@ -27,7 +27,7 @@ agents:
     type: workflow
     redis_port: 6379
     api_port: 8080 # Only needed for workflows, defaults to 8080 if not filled
-    workflow_file: workflows/example_workflow.py
+    workflow_file: workflow/example_workflow.py
     provider: EC2
     instance_type: t3.micro
 

examples/portfolio/agents/advisor_agent.py

@@ -2,7 +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). Configure with env vars:
+# (Converse API), called via ventis.llm.bedrock so token/cost telemetry gets
+# recorded onto this execution's future:<future_id>:metrics hash. Configure
+# with env vars:
 #   BEDROCK_MODEL_ID  (default: meta.llama3-8b-instruct-v1:0)
 #   AWS_REGION        (default: us-east-1)
 #
@@ -13,6 +15,11 @@
 
 import os
 
+try:
+    from ventis.llm.bedrock import call_bedrock
+except ImportError:
+    from bedrock import call_bedrock
+
 
 class AdvisorAgent(object):
     def __init__(self):
@@ -26,13 +33,11 @@ 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:
-            import boto3
-
-            client = boto3.client("bedrock-runtime", region_name=self.region)
-            response = client.converse(
-                modelId=self.model_id,
+            response = call_bedrock(
+                model_id=self.model_id,
                 messages=[{"role": "user", "content": [{"text": prompt}]}],
-                inferenceConfig={"maxTokens": 400, "temperature": 0.2},
+                inference_config={"maxTokens": 400, "temperature": 0.2},
+                region=self.region,
             )
             return response["output"]["message"]["content"][0]["text"]
         except Exception as e:

examples/portfolio/agents/intent_agent.py

@@ -0,0 +1,127 @@
+# Intent Agent
+#
+# Stage 0. Turns a free-text portfolio request into the structured input the
+# rest of the pipeline needs:
+#
+#   "Analyze 40% Apple, 35% Microsoft and 25% Nvidia over the last 6 months"
+#      ->  {"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>:metrics hash. Configure with env vars:
+#   BEDROCK_MODEL_ID  (default: meta.llama3-8b-instruct-v1:0)
+#   AWS_REGION        (default: us-east-1)
+#
+# If the LLM is unavailable or returns unparseable output, parse() raises:
+# there is no fallback, the request fails loudly rather than guessing at the
+# holdings. Weights are renormalized to 1.0.
+#
+# Resource profile: cheap CPU, single call per request, on the critical path
+# before the fan-out.
+
+import os
+import re
+import json
+
+try:
+    from ventis.llm.bedrock import call_bedrock
+except ImportError:
+    from bedrock import call_bedrock
+
+DEFAULT_LOOKBACK_DAYS = 365
+
+
+class IntentAgent(object):
+    def __init__(self):
+        self.tools = [self.parse]
+        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")
+
+    def parse(self, query: str) -> dict:
+        """Parse a natural-language portfolio request into holdings + lookback."""
+        response = call_bedrock(
+            model_id=self.model_id,
+            messages=[{"role": "user", "content": [{"text": self._build_prompt(query)}]}],
+            inference_config={"maxTokens": 300, "temperature": 0.0},
+            region=self.region,
+        )
+        text = response["output"]["message"]["content"][0]["text"]
+        if not text:
+            raise ValueError("IntentAgent: LLM returned no output for the request.")
+
+        parsed = self._extract_json(text)
+        if parsed is None:
+            raise ValueError(
+                f"IntentAgent: could not parse holdings from LLM output: {text!r}"
+            )
+
+        result = self._sanitize(parsed)
+        if not result["holdings"]:
+            raise ValueError(
+                f"IntentAgent: no valid holdings found in request: {query!r}"
+            )
+        return result
+
+    def _build_prompt(self, query: str) -> str:
+        return (
+            "You convert a plain-English portfolio request into JSON. Return ONLY "
+            "a JSON object, no prose, with exactly two keys:\n"
+            '  "holdings": an object mapping stock TICKER symbols (uppercase) to '
+            "their weight as a decimal fraction (weights should sum to about 1.0), and\n"
+            '  "lookback_days": an integer number of calendar days for the analysis '
+            f"window (default {DEFAULT_LOOKBACK_DAYS} if unspecified; 1 month = 30 "
+            "days, 1 year = 365 days).\n"
+            "Resolve company names to their ticker (Apple->AAPL, Microsoft->MSFT, "
+            "Nvidia->NVDA, etc.). If weights are given as percentages, convert to "
+            "fractions. If a holding has no explicit weight, split the remainder "
+            "equally among the unweighted holdings.\n\n"
+            f'Request: "{query}"\n\n'
+            "JSON:"
+        )
+
+    def _extract_json(self, text: str):
+        """Pull the first JSON object out of the model's response text."""
+        # Models sometimes wrap the JSON in prose or code fences; grab the
+        # outermost {...} span.
+        match = re.search(r"\{.*\}", text, re.DOTALL)
+        if not match:
+            return None
+        try:
+            return json.loads(match.group(0))
+        except (ValueError, TypeError):
+            return None
+
+    def _sanitize(self, parsed: dict) -> dict:
+        """Validate types and renormalize weights to sum to 1.0."""
+        raw = (parsed or {}).get("holdings") or {}
+        holdings = {}
+        for ticker, weight in raw.items():
+            try:
+                w = float(weight)
+            except (ValueError, TypeError):
+                continue
+            if w > 0:
+                holdings[str(ticker).upper()] = w
+
+        total = sum(holdings.values())
+        if total > 0:
+            holdings = {t: round(w / total, 4) for t, w in holdings.items()}
+
+        try:
+            lookback = int((parsed or {}).get("lookback_days", DEFAULT_LOOKBACK_DAYS))
+        except (ValueError, TypeError):
+            lookback = DEFAULT_LOOKBACK_DAYS
+        if lookback <= 0:
+            lookback = DEFAULT_LOOKBACK_DAYS
+
+        return {"holdings": holdings, "lookback_days": lookback}
+
+
+if __name__ == "__main__":
+    agent = IntentAgent()
+    print(agent.parse(
+        query="Analyze 40% Apple, 35% Microsoft and 25% Nvidia over the last 6 months"
+    ))

examples/portfolio/agents/intent_agent.yaml

@@ -0,0 +1,10 @@
+agent:
+  name: IntentAgent
+  functions:
+    - name: parse
+      description: Parse a natural-language portfolio request into holdings + lookback.
+      arguments:
+        - name: query
+          type: str
+      returns:
+        type: dict

examples/portfolio/agents/metrics_agent.py

@@ -9,6 +9,7 @@
 #
 # Resource profile: cheap CPU, high fan-out — one compute() call per holding.
 
+import json
 import math
 
 from price_agent import PriceAgent
@@ -23,9 +24,12 @@ def __init__(self):
 
     def compute(self, ticker: str, lookback_days: int = 365) -> dict:
         """Compute return/volatility/Sharpe/drawdown metrics for one ticker."""
-        history = self.price.get_history(
-            ticker=ticker, lookback_days=lookback_days
-        ).value()
+        # get_history() returns a dict, but a Future's .value() only ever gives back
+        # the raw string ventis stored in Redis -- it never auto-deserializes
+        # non-str return types, so the JSON has to be parsed back out here.
+        history = json.loads(
+            self.price.get_history(ticker=ticker, lookback_days=lookback_days).value()
+        )
         closes = history.get("closes", [])
 
         if len(closes) < 2:

examples/portfolio/agents/price_agent.py

@@ -18,11 +18,21 @@ def __init__(self):
     def get_history(self, ticker: str, lookback_days: int = 365) -> dict:
         """Fetch daily closing prices for a ticker over the lookback window."""
         try:
+            import math
+
             import yfinance as yf
 
             hist = yf.Ticker(ticker).history(period=f"{lookback_days}d")
-            closes = [float(c) for c in hist["Close"].tolist()]
-            dates = [str(d.date()) for d in hist.index]
+            raw_closes = hist["Close"].tolist()
+            # Yahoo sometimes appends the newest trading day before its close has
+            # settled, leaving that row's Close as NaN -- drop it rather than let
+            # NaN poison every downstream metric.
+            closes = [float(c) for c in raw_closes if not math.isnan(c)]
+            dates = [
+                str(d.date())
+                for d, c in zip(hist.index, raw_closes)
+                if not math.isnan(c)
+            ]
             if closes:
                 return {
                     "ticker": ticker,

examples/portfolio/config/global_controller.yaml

@@ -6,63 +6,76 @@
 # reflect each stage's real cost so the scheduler has placement decisions to make.
 
 agents:
-  # Stage 0: price history fetch. Network/IO-bound, cheap CPU. Called by
+  # Stage 0: parse the free-text request into structured holdings + lookback
+  # window (calls Bedrock directly via ventis.llm.bedrock). Cheap CPU, one
+  # call per request, on the critical path before the fan-out.
+  - name: IntentAgent
+    redis_port: 6379
+    replicas: 1
+    resources:
+      cpu: 1
+      memory: 256
+    entrypoint: agents/intent_agent.py
+    provider: EC2
+    instance_type: t3.micro
+
+  # Stage 0b: price history fetch. Network/IO-bound, cheap CPU. Called by
   # MetricsAgent, one fetch per ticker — high fan-out, so multiple replicas.
   - name: PriceAgent
-    host: localhost
-    port: 8071
     redis_port: 6379
-    replicas: 2
+    replicas: 1
     resources:
       cpu: 1
       memory: 256
     entrypoint: agents/price_agent.py
+    provider: EC2
+    instance_type: t3.micro
 
   # Stage 1: per-ticker metrics. Cheap CPU, high fan-out (one call per holding).
   # Calls PriceAgent remotely, then computes return/vol/Sharpe/drawdown.
   - name: MetricsAgent
-    host: localhost
-    port: 8072
     redis_port: 6379
-    replicas: 2
+    replicas: 1
     resources:
       cpu: 1
       memory: 256
     entrypoint: agents/metrics_agent.py
+    provider: EC2
+    instance_type: t3.micro
 
   # Stage 2: portfolio-level risk aggregation. Single call per request; needs
   # every ticker's metrics (the barrier).
   - name: RiskAgent
-    host: localhost
-    port: 8073
     redis_port: 6379
     replicas: 1
     resources:
       cpu: 1
       memory: 256
     entrypoint: agents/risk_agent.py
+    provider: EC2
+    instance_type: t3.micro
 
   # Stage 3: LLM briefing via Bedrock. On the critical path, one call per
   # request.
   - name: AdvisorAgent
-    host: localhost
-    port: 8074
     redis_port: 6379
     replicas: 1
     resources:
       cpu: 1
       memory: 512
     entrypoint: agents/advisor_agent.py
+    provider: EC2
+    instance_type: t3.micro
 
   # The workflow, exposed as a REST API.
   - name: Workflow
-    host: localhost
-    port: 8070             # LC gRPC port
     type: workflow
     api_port: 8080         # Flask REST API port
     redis_port: 6379
     replicas: 1
-    workflow_file: workflows/portfolio_workflow.py
+    workflow_file: workflow/portfolio_workflow.py
+    provider: EC2
+    instance_type: t3.micro
 
 # Polling interval in seconds
 poll_interval: 5

examples/portfolio/config/policy.yaml

@@ -14,6 +14,7 @@ rules:
   - match: {}
     access:
       - Workflow
+      - IntentAgent
       - PriceAgent
       - MetricsAgent
       - RiskAgent