Merge branch 'develop' into feature/implement-hamilton-framework-with-sdlc-11-28-51

Author: 2-Coatl

scripts/coding/ai/examples/hamilton_llm/__init__.py

@@ -0,0 +1,12 @@
+"""Hamilton-inspired LLM pipeline example for the IACT project."""
+
+from . import dataflow
+from .driver import HamiltonDriver, MissingDependencyError
+from .llm_client import MockLLMClient
+
+__all__ = [
+    "dataflow",
+    "HamiltonDriver",
+    "MissingDependencyError",
+    "MockLLMClient",
+]

scripts/coding/ai/examples/hamilton_llm/dataflow.py

@@ -0,0 +1,128 @@
+"""Declarative dataflow modeling the Data → Prompt → LLM → $ pipeline.
+
+The module captures the pace differences between aplicaciones ML tradicionales y
+aplicaciones LLM, destacando que ambas requieren habilidades fuertes de
+ingeniería de software. Cada función sigue el paradigma Hamilton: el nombre es
+el output y los argumentos son las dependencias explícitas.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List
+
+from .llm_client import MockLLMClient
+
+PACE_OF_DEVELOPMENT: Dict[str, List[str]] = {
+    "traditional_ml": [
+        "Idea & Data/Resources",
+        "Design",
+        "Development/Prototype",
+        "Model Development",
+        "Getting to Production",
+        "Operations",
+        "Maintenance & Business Value",
+    ],
+    "llm_apps": [
+        "Idea & Data/Resources",
+        "Design",
+        "Development/Prototype",
+        "Prompt / Model Development",
+        "Getting to Production",
+        "Operations",
+        "Maintenance & Business Value",
+    ],
+}
+
+DATAFLOW_LABEL = "Data → Prompt → LLM → $"
+
+
+def pace_of_development() -> Dict[str, List[str]]:
+    """Return the canonical ordering of fases para ML tradicional y apps LLM."""
+
+    return PACE_OF_DEVELOPMENT
+
+
+def prompt_template(
+    idea: str,
+    domain_data: Dict[str, str],
+    pace_of_development: Dict[str, List[str]],
+) -> str:
+    """Create a template that contrasta los ritmos y exige prácticas SWE."""
+
+    traditional = " → ".join(pace_of_development["traditional_ml"])
+    llm = " → ".join(pace_of_development["llm_apps"])
+    return (
+        "You are designing a Hamilton micro-orchestration experiment.\n"
+        f"Traditional ML pace: {traditional}.\n"
+        f"LLM app pace: {llm}.\n"
+        "Explain how strong SWE practices (testing, modularity, reuse, portability)\n"
+        "keep the system resilient while iterating quickly.\n"
+        f"Business domain: {domain_data['business_process']} with UI {domain_data['ui']}.\n"
+        f"Primary data assets: {domain_data['data']}.\n"
+        f"Goal: deliver {idea} using Hamilton declarative functions.\n"
+    )
+
+
+def llm_prompt(prompt_template: str, edge_cases: List[str]) -> str:
+    """Combine template with guardrails against edge cases y prompt injection."""
+
+    formatted_edge_cases = ", ".join(edge_cases)
+    return (
+        f"{prompt_template}"
+        "Consider the following edge cases explicitly: "
+        f"{formatted_edge_cases}.\n"
+        "Detail the pipeline as Data → Prompt → LLM → $, highlighting how guardrails\n"
+        "prevent prompt injection and balance evaluation with GPU cost awareness."
+    )
+
+
+def llm_response(llm_prompt: str, llm_client: MockLLMClient) -> str:
+    """Obtain respuesta determinística del cliente LLM simulado."""
+
+    return llm_client.complete(llm_prompt)
+
+
+def prompt_token_estimate(llm_prompt: str, edge_cases: List[str]) -> int:
+    """Estimate token count con amortiguador para cobertura de edge cases."""
+
+    narrative_tokens = len(llm_prompt.split())
+    scaled_tokens = round(narrative_tokens * 0.75)
+    guardrail_tokens = len(edge_cases) * 3
+    return max(scaled_tokens + guardrail_tokens, 120)
+
+
+def business_value(
+    llm_response: str,
+    pace_of_development: Dict[str, List[str]],
+) -> Dict[str, Any]:
+    """Empaquetar plan de acción y el contexto de ritmo de desarrollo."""
+
+    return {
+        "llm_plan": llm_response,
+        "pace": pace_of_development,
+        "next_step": "Prototype with guarded prompts",
+    }
+
+
+def cost_estimate(
+    prompt_token_estimate: int,
+    pricing_policy: Dict[str, float],
+) -> float:
+    """Calcular costo esperado usando tarifa por 1K tokens y factor de seguridad."""
+
+    price = pricing_policy["price_per_1k_tokens"]
+    safety = pricing_policy.get("safety_multiplier", 1.0)
+    return round((prompt_token_estimate / 1000) * price * safety, 6)
+
+
+__all__ = [
+    "PACE_OF_DEVELOPMENT",
+    "DATAFLOW_LABEL",
+    "pace_of_development",
+    "prompt_template",
+    "llm_prompt",
+    "llm_response",
+    "prompt_token_estimate",
+    "business_value",
+    "cost_estimate",
+]

scripts/coding/ai/examples/hamilton_llm/driver.py

@@ -0,0 +1,79 @@
+"""Minimal Hamilton-like driver for executing declarative dataflows.
+
+The real Hamilton framework provides a rich micro-orchestration engine. For the
+purposes of the repository we build a tiny subset that resolves dependencies by
+function name and executes only the nodes required to produce requested targets.
+"""
+
+from __future__ import annotations
+
+import inspect
+from types import ModuleType
+from typing import Any, Dict, Iterable, Mapping, Sequence
+
+
+class MissingDependencyError(RuntimeError):
+    """Raised when a dependency required by a node is not available."""
+
+
+class HamiltonDriver:
+    """Execute declarative functions registered from one or more modules.
+
+    Functions are registered by name and resolved lazily. Inputs provided via
+    ``execute`` act as seed values, mirroring Hamilton's configuration
+    dictionary. Each execution resets the cache and produces a log of executed
+    nodes so tests can assert on evaluation order.
+    """
+
+    def __init__(self, modules: Iterable[ModuleType]):
+        self._functions: Dict[str, Any] = {}
+        self.execution_log: list[str] = []
+        for module in modules:
+            self._register_module(module)
+
+    def _register_module(self, module: ModuleType) -> None:
+        for name, candidate in vars(module).items():
+            if inspect.isfunction(candidate):
+                self._functions[name] = candidate
+
+    def execute(self, targets: Sequence[str], inputs: Mapping[str, Any]) -> Dict[str, Any]:
+        cache: Dict[str, Any] = {}
+        context: Dict[str, Any] = dict(inputs)
+        self.execution_log = []
+
+        def resolve(name: str) -> Any:
+            if name in cache:
+                return cache[name]
+            if name in context:
+                return context[name]
+
+            func = self._functions.get(name)
+            if func is None:
+                raise MissingDependencyError(f"No data or function available for '{name}'")
+
+            signature = inspect.signature(func)
+            kwargs: Dict[str, Any] = {}
+            for parameter in signature.parameters.values():
+                if parameter.kind in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD):
+                    raise MissingDependencyError(
+                        f"Unsupported parameter kind for '{func.__name__}': {parameter.kind}"
+                    )
+                dependency_name = parameter.name
+                try:
+                    kwargs[dependency_name] = resolve(dependency_name)
+                except MissingDependencyError as exc:  # pragma: no cover - rephrase message
+                    raise MissingDependencyError(
+                        f"Function '{func.__name__}' requires missing dependency '{dependency_name}'"
+                    ) from exc
+
+            value = func(**kwargs)
+            cache[name] = value
+            context[name] = value
+            self.execution_log.append(name)
+            return value
+
+        results = {target: resolve(target) for target in targets}
+        return results
+
+
+__all__ = ["HamiltonDriver", "MissingDependencyError"]

scripts/coding/ai/examples/hamilton_llm/llm_client.py

@@ -0,0 +1,28 @@
+"""Deterministic mock client emulating an LLM completion API."""
+
+from __future__ import annotations
+
+from typing import Mapping
+
+
+class MockLLMClient:
+    """Return canned responses y exponer tarifa para estimar costos."""
+
+    def __init__(self, price_per_1k_tokens: float, response_catalog: Mapping[str, str]):
+        self.price_per_1k_tokens = price_per_1k_tokens
+        self._response_catalog = dict(response_catalog)
+
+    def complete(self, prompt: str) -> str:
+        """Return the first response cuyo identificador esté contenido en el prompt."""
+
+        lower_prompt = prompt.lower()
+        for key, response in self._response_catalog.items():
+            if key.lower() in lower_prompt:
+                return response
+        return self._response_catalog.get(
+            "__default__",
+            "Document modular functions, validate with pytest and guard against prompt injection.",
+        )
+
+
+__all__ = ["MockLLMClient"]

scripts/coding/tests/ai/examples/__init__.py

@@ -0,0 +1,115 @@
+"""Tests for the Hamilton-style LLM dataflow example."""
+
+import pytest
+
+from infrastructure.workspace.hamilton_llm import dataflow
+from infrastructure.workspace.hamilton_llm import driver as mini_driver
+from infrastructure.workspace.hamilton_llm.driver import MissingDependencyError
+from infrastructure.workspace.hamilton_llm.llm_client import MockLLMClient
+
+
+def test_pace_of_development_metadata_matches_expected_sequence():
+    """Validate that the module exposes the canonical pacing differences described in the slides."""
+    assert dataflow.PACE_OF_DEVELOPMENT["traditional_ml"] == [
+        "Idea & Data/Resources",
+        "Design",
+        "Development/Prototype",
+        "Model Development",
+        "Getting to Production",
+        "Operations",
+        "Maintenance & Business Value",
+    ]
+    assert dataflow.PACE_OF_DEVELOPMENT["llm_apps"] == [
+        "Idea & Data/Resources",
+        "Design",
+        "Development/Prototype",
+        "Prompt / Model Development",
+        "Getting to Production",
+        "Operations",
+        "Maintenance & Business Value",
+    ]
+
+
+def test_hamilton_builder_executes_llm_business_flow():
+    """End-to-end execution should transform data into a business value package and cost estimate."""
+    driver = (
+        mini_driver.Builder()
+        .with_modules(dataflow)
+        .with_config({"pricing_policy": {"price_per_1k_tokens": 0.4, "safety_multiplier": 1.15}})
+        .with_adapters(mini_driver.DictResult())
+        .build()
+    )
+    mock_client = MockLLMClient(
+        price_per_1k_tokens=0.4,
+        response_catalog={
+            "Data → Prompt → LLM → $": "Use Hamilton declarative functions to keep prompts versioned and guarded against injection."
+        },
+    )
+
+    inputs = {
+        "idea": "AI copilots for compliance analysts",
+        "domain_data": {
+            "data": "archived compliance tickets",
+            "ui": "browser extension",
+            "business_process": "regulatory audit",
+        },
+        "edge_cases": [
+            "Input state space",
+            "Guard against prompt injection",
+            "Domain expertise",
+            "Evaluation",
+            "Cost/GPUs",
+        ],
+        "llm_client": mock_client,
+    }
+
+    result = driver.execute(["business_value", "cost_estimate"], inputs)
+
+    assert result["business_value"]["llm_plan"].startswith("Use Hamilton declarative functions")
+    assert result["business_value"]["next_step"] == "Prototype with guarded prompts"
+    assert pytest.approx(result["cost_estimate"], rel=1e-3) == 0.0552
+
+    executed = driver.execution_log
+    assert executed[-1] == "cost_estimate"
+    assert "llm_response" in executed
+    assert executed.index("business_value") < executed.index("cost_estimate")
+
+
+def test_driver_reports_missing_inputs():
+    driver = mini_driver.Builder().with_modules(dataflow).build()
+    mock_client = MockLLMClient(price_per_1k_tokens=0.5, response_catalog={})
+
+    with pytest.raises(MissingDependencyError) as exc:
+        driver.execute(
+            ["cost_estimate"],
+            {
+                "idea": "Guardrails demo",
+                "domain_data": {"data": "logs", "ui": "cli", "business_process": "ops"},
+                "edge_cases": ["Injection"],
+                "llm_client": mock_client,
+            },
+        )
+
+    assert "pricing_policy" in str(exc.value)
+
+
+def test_builder_requires_modules_before_building():
+    with pytest.raises(ValueError) as exc:
+        mini_driver.Builder().build()
+
+    assert "modules" in str(exc.value).lower()
+
+
+def test_custom_adapter_transforms_execution_result():
+    class KeysAdapter:
+        def __call__(self, results):
+            return tuple(sorted(results))
+
+    driver = (
+        mini_driver.Builder()
+        .with_modules(dataflow)
+        .with_adapters(KeysAdapter())
+        .build()
+    )
+
+    assert driver.execute(["pace_of_development"], {}) == ("pace_of_development",)