Add post-processing pipeline for carbon impact, pricing, and equivalences. Introduce orchestrator and context management for enhanced metadata handling in agent service.

config/settings.py

@@ -32,6 +32,39 @@ class Settings(BaseSettings):
     supabase_match_fn: str = "match_documents"
     rag_top_k: int = 5
     
+    # Post-processing pipeline configuration
+    postprocessors_enabled: list[str] = [
+        "carbon_impact",
+        "pricing",
+        "equivalences",
+    ]
+    
+    currency: str = "USD"
+
+    # Pricing per 1,000,000 tokens (input/output)
+    pricing: dict = {
+        # Complete or override in .env via nested JSON if needed
+        "mistral-medium-latest": {"input_per_1m": 0.40, "output_per_1m": 2.00},
+        "mistral-small-latest": {"input_per_1m": 0.10, "output_per_1m": 0.30},
+        "mistral-large-latest": {"input_per_1m": 2.00, "output_per_1m": 6.00},
+        "magistral-small-latest": {"input_per_1m": 0.50, "output_per_1m": 1.50},
+        "magistral-medium-latest": {"input_per_1m": 2.00, "output_per_1m": 5.00},
+    }
+
+    # Equivalence ratios using kgCO2eq as input
+    equivalence_ratios: dict = {
+        # https://impactco2.fr/outils/comparateur
+        # Ratios expressed as UNITS per kgCO2eq (invert of kgCO2 per unit)
+        # smartphone: 85.9 kgCO2 per unit → 1 / 85.9 ≈ 0.011643 smartphones per kgCO2
+        "smartphone_per_kgCO2eq": 0.011643,
+        # car (km): 219 g CO2 per km → 0.219 kg per km → 1 / 0.219 ≈ 4.566210 km per kgCO2
+        "car_km_per_kgCO2eq": 4.566210,
+        # tgv (km): 2.93 g CO2 per km → 0.00293 kg per km → 1 / 0.00293 ≈ 341.296928 km per kgCO2
+        "tgv_km_per_kgCO2eq": 341.296928,
+        # water (l): 321 g CO2 per liter → 0.321 kg per liter → 1 / 0.321 ≈ 3.115265 liters per kgCO2
+        "water_l_per_kgCO2eq": 3.115265,
+    }
+    
     model_config = SettingsConfigDict(
         env_file=".env",
         env_file_encoding="utf-8",

services/agent_service.py

@@ -6,7 +6,8 @@ from langchain_core.language_models.chat_models import BaseChatModel
 from domain.enums import ModelName, AgentType
 from .llm_service import llm_service
 from .agent_registry import agent_registry
-from .impact_service import trace_llm_impact
+from services.postprocessing.registry import build_orchestrator
+from services.postprocessing.context import RunContext
 
 
 class AgentService:
@@ -62,21 +63,40 @@ class AgentService:
         # Prepare messages
         messages = self._prepare_messages(message, conversation_history)
         
-        # Execute graph
+        # Execute graph with latency
+        start_time = time.time()
         result = await graph.ainvoke({"messages": messages})
+        latency_s = time.time() - start_time
         
         # Extract response
         response_message = result["messages"][-1]
         response_content = response_message.content
 
+        # Prepare metadata and run post-processing pipeline
+        usage = getattr(response_message, "usage_metadata", None) or {}
+        usage_totals = self._normalize_usage(usage)
+        usage_by_model = {model_name.value: usage_totals}
+
+        ctx = RunContext(
+            provider=model_name.provider.value,
+            model=model_name.value,
+            usage_totals=usage_totals,
+            usage_by_model=usage_by_model,
+            latency_s=latency_s,
+        )
+        build_orchestrator().run(ctx)
+
+        base_metadata: Dict[str, Any] = {
+            "message_count": len(result["messages"]),
+        }
+        base_metadata.update(ctx.metadata_out)
+
         return {
             "response": response_content,
             "model": model_name.value,
             "agent_type": agent_type.value,
-            "usage": getattr(response_message, "usage_metadata", None),
-            "metadata": {
-                "message_count": len(result["messages"])
-            }
+            "usage": usage,
+            "metadata": base_metadata,
         }
     
     async def stream(
@@ -205,18 +225,16 @@ class AgentService:
                     "documents": documents
                 }
         
-        # Compute latency and emissions for the final chunk
+        # Compute latency and run post-processing pipeline for the final chunk
         latency_s = time.time() - start_time
-        emissions_kg = None
-        try:
-            emissions_kg = trace_llm_impact(
-                provider=model_name.provider.value,
-                model=model_name.value,
-                usage=usage_totals or {},
-                latency=latency_s,
-            )
-        except Exception:
-            emissions_kg = None
+        ctx = RunContext(
+            provider=model_name.provider.value,
+            model=model_name.value,
+            usage_totals=usage_totals,
+            usage_by_model=usage_by_model,
+            latency_s=latency_s,
+        )
+        build_orchestrator().run(ctx)
 
         # Send final chunk
         yield {
@@ -228,7 +246,7 @@ class AgentService:
                 "usage": usage_totals,
                 "usage_by_model": usage_by_model,
                 "latency_s": latency_s,
-                **({"emissions_kgCO2eq": emissions_kg, "emissions_gCO2eq": emissions_kg * 1000.0} if emissions_kg is not None else {})
+                **ctx.metadata_out
             },
             "documents": documents
         }

services/postprocessing/base.py

@@ -0,0 +1,25 @@
+"""Post-processing pipeline base interfaces and orchestrator."""
+from typing import Protocol, List
+from .context import RunContext
+
+
+class PostProcessor(Protocol):
+    name: str
+
+    def process(self, ctx: RunContext) -> None:
+        ...
+
+
+class PostProcessingOrchestrator:
+    def __init__(self, processors: List[PostProcessor]):
+        self._processors = processors
+
+    def run(self, ctx: RunContext) -> None:
+        for processor in self._processors:
+            try:
+                processor.process(ctx)
+            except Exception:
+                # Best effort: don't break the response if a processor fails
+                continue
+
+

services/postprocessing/context.py

@@ -0,0 +1,15 @@
+"""Shared context for post-processing pipeline."""
+from dataclasses import dataclass, field
+from typing import Dict, Any
+
+
+@dataclass
+class RunContext:
+    provider: str
+    model: str
+    usage_totals: Dict[str, int]
+    usage_by_model: Dict[str, Dict[str, int]]
+    latency_s: float
+    metadata_out: Dict[str, Any] = field(default_factory=dict)
+
+

services/postprocessing/processors/carbon_impact.py

@@ -0,0 +1,15 @@
+"""Carbon impact post-processor using ecologits."""
+from services.impact_service import trace_llm_impact
+from ..context import RunContext
+
+
+class CarbonImpactProcessor:
+    name = "carbon_impact"
+
+    def process(self, ctx: RunContext) -> None:
+        kg = trace_llm_impact(ctx.provider, ctx.model, ctx.usage_totals, ctx.latency_s)
+        if kg is not None:
+            ctx.metadata_out["emissions_kgCO2eq"] = kg
+            ctx.metadata_out["emissions_gCO2eq"] = kg * 1000.0
+
+

services/postprocessing/processors/equivalences.py

@@ -0,0 +1,25 @@
+"""Equivalences post-processor mapping kgCO2eq to concrete equivalents."""
+from config import settings
+from ..context import RunContext
+
+
+class EquivalencesProcessor:
+    name = "equivalences"
+
+    def process(self, ctx: RunContext) -> None:
+        kg = ctx.metadata_out.get("emissions_kgCO2eq")
+        if kg is None:
+            return
+        ratios = getattr(settings, "equivalence_ratios", {})
+        try:
+            eq = {
+                "water_liters": round(kg * float(ratios.get("water_l_per_kgCO2eq", 0)), 4),
+                "car_km": round(kg * float(ratios.get("car_km_per_kgCO2eq", 0)), 4),
+                "tgv_km": round(kg * float(ratios.get("tgv_km_per_kgCO2eq", 0)), 4),
+                "smartphone": round(kg * float(ratios.get("smartphone_per_kgCO2eq", 0)), 4),
+            }
+        except Exception:
+            return
+        ctx.metadata_out["equivalences"] = eq
+
+

services/postprocessing/processors/pricing.py

@@ -0,0 +1,45 @@
+"""Pricing post-processor computing costs per model and totals."""
+from config import settings
+from ..context import RunContext
+
+
+class PricingProcessor:
+    name = "pricing"
+
+    def process(self, ctx: RunContext) -> None:
+        currency = getattr(settings, "currency", "EUR")
+        pricing_cfg = getattr(settings, "pricing", {})
+
+        by_model = {}
+        total_cost = 0.0
+
+        for model_id, usage in ctx.usage_by_model.items():
+            cfg = pricing_cfg.get(model_id) or {}
+            # Prefer per-million token pricing; fallback to per-1k if provided
+            in_rate_1m = cfg.get("input_per_1m")
+            out_rate_1m = cfg.get("output_per_1m")
+            if in_rate_1m is None and cfg.get("input_per_1k") is not None:
+                in_rate_1m = float(cfg.get("input_per_1k")) * 1000.0
+            if out_rate_1m is None and cfg.get("output_per_1k") is not None:
+                out_rate_1m = float(cfg.get("output_per_1k")) * 1000.0
+
+            in_rate_1m = float(in_rate_1m or 0)
+            out_rate_1m = float(out_rate_1m or 0)
+
+            cost_in = (usage.get("input_tokens", 0) / 1_000_000.0) * in_rate_1m
+            cost_out = (usage.get("output_tokens", 0) / 1_000_000.0) * out_rate_1m
+            model_total = cost_in + cost_out
+            by_model[model_id] = {
+                "input": round(cost_in, 6),
+                "output": round(cost_out, 6),
+                "total": round(model_total, 6),
+            }
+            total_cost += model_total
+
+        ctx.metadata_out["pricing"] = {
+            "currency": currency,
+            "total_cost": round(total_cost, 6),
+            "by_model": by_model,
+        }
+
+

services/postprocessing/registry.py

@@ -0,0 +1,25 @@
+"""Registry for building the post-processing orchestrator."""
+from config import settings
+from .base import PostProcessingOrchestrator
+from .processors.carbon_impact import CarbonImpactProcessor
+from .processors.pricing import PricingProcessor
+from .processors.equivalences import EquivalencesProcessor
+
+
+ALL = {
+    "carbon_impact": CarbonImpactProcessor(),
+    "pricing": PricingProcessor(),
+    "equivalences": EquivalencesProcessor(),
+}
+
+
+def build_orchestrator() -> PostProcessingOrchestrator:
+    enabled = getattr(settings, "postprocessors_enabled", [
+        "carbon_impact",
+        "pricing",
+        "equivalences",
+    ])
+    processors = [ALL[name] for name in enabled if name in ALL]
+    return PostProcessingOrchestrator(processors)
+
+