feat(v1): Added pipeline.py, compiler.py

requirements.txt

@@ -0,0 +1,31 @@
+# Farmer.Chat Backend Dependencies
+
+# Web Framework
+fastapi==0.109.0
+uvicorn[standard]==0.27.0
+python-multipart==0.0.6
+
+# OpenAI
+openai==1.10.0
+
+# Async HTTP
+aiohttp==3.9.1
+aiofiles==23.2.1
+
+# Data Processing
+pandas==2.1.4
+numpy==1.26.3
+xarray==2023.12.0
+netCDF4==1.6.5
+
+# PDF Generation
+reportlab==4.0.9
+pillow==10.2.0
+
+# Utilities
+python-dotenv==1.0.0
+pydantic==2.5.3
+pydantic-settings==2.1.0
+
+# Requests (for sync operations)
+requests==2.31.0

src/compiler.py

@@ -0,0 +1,57 @@
+"""
+Stage 3: Response Compiler - Data Fusion
+"""
+
+from typing import Dict, Any, List
+
+
+class ResponseCompiler:
+    """Stage 3: Compile results from multiple servers"""
+
+    def compile(self, raw_results: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Merge results into structured format
+        
+        Args:
+            raw_results: Dictionary containing results from MCPExecutor
+                {
+                    "results": {
+                        "weather": {"status": "success", "data": {...}},
+                        "soil_properties": {"status": "success", "data": {...}},
+                        ...
+                    },
+                    "execution_time_seconds": 3.5
+                }
+        
+        Returns:
+            {
+                "successful_servers": List[str],
+                "failed_servers": List[dict],
+                "data": Dict[str, Any],
+                "execution_time": float,
+                "completeness": str
+            }
+        """
+        results_dict = raw_results.get("results", {})
+
+        successful = []
+        failed = []
+        compiled_data = {}
+
+        for server_name, result in results_dict.items():
+            if result.get("status") == "success":
+                successful.append(server_name)
+                compiled_data[server_name] = result.get("data", {})
+            else:
+                failed.append({
+                    "server": server_name,
+                    "error": result.get("error", "Unknown error")
+                })
+
+        return {
+            "successful_servers": successful,
+            "failed_servers": failed,
+            "data": compiled_data,
+            "execution_time": raw_results.get("execution_time_seconds", 0),
+            "completeness": f"{len(successful)}/{len(results_dict)} servers"
+        }

src/pipeline.py

@@ -0,0 +1,93 @@
+"""
+Complete Multi-Stage MCP Pipeline
+Orchestrates: Router → Executor → Compiler → Translator
+"""
+
+import time
+from typing import Dict, Any
+from openai import OpenAI
+
+from .router import QueryRouter
+from .executor import MCPExecutor, MCP_SERVER_REGISTRY
+from .compiler import ResponseCompiler
+from .translator import FarmerTranslator
+
+
+class FarmerChatPipeline:
+    """Complete multi-stage MCP pipeline"""
+
+    def __init__(self, openai_client: OpenAI, location: Dict[str, Any]):
+        self.location = location
+        self.router = QueryRouter(openai_client, MCP_SERVER_REGISTRY)
+        self.executor = MCPExecutor()
+        self.compiler = ResponseCompiler()
+        self.translator = FarmerTranslator(openai_client)
+
+    async def process_query(self, query: str, verbose: bool = False) -> Dict[str, Any]:
+        """
+        Process farmer query through complete pipeline
+        
+        Returns:
+            {
+                "query": str,
+                "routing": dict,
+                "compiled_data": dict,
+                "advice": str,
+                "pipeline_time_seconds": float
+            }
+        """
+        if verbose:
+            print(f"\n🌾 Processing: {query}")
+            print(f"📍 Location: {self.location['name']}")
+
+        pipeline_start = time.time()
+
+        # STAGE 1: Query Routing
+        if verbose:
+            print("🎯 Stage 1: Routing...")
+        
+        routing = self.router.route(query, self.location)
+        
+        if verbose:
+            print(f"   → Servers: {', '.join(routing['required_servers'])}")
+
+        # STAGE 2: MCP Execution (Parallel)
+        if verbose:
+            print("⚙️ Stage 2: Executing MCP servers...")
+        
+        raw_results = await self.executor.execute_parallel(
+            routing['required_servers'],
+            self.location['lat'],
+            self.location['lon']
+        )
+        
+        if verbose:
+            print(f"   → Completed in {raw_results['execution_time_seconds']}s")
+
+        # STAGE 3: Response Compilation
+        if verbose:
+            print("🔗 Stage 3: Compiling results...")
+        
+        compiled = self.compiler.compile(raw_results)
+        
+        if verbose:
+            print(f"   → {compiled['completeness']}")
+
+        # STAGE 4: Farmer Translation
+        if verbose:
+            print("🌾 Stage 4: Generating advice...")
+        
+        farmer_advice = self.translator.translate(query, compiled, self.location)
+
+        pipeline_time = time.time() - pipeline_start
+
+        if verbose:
+            print(f"✅ Complete! Total: {pipeline_time:.2f}s\n")
+
+        return {
+            "query": query,
+            "routing": routing,
+            "compiled_data": compiled,
+            "advice": farmer_advice,
+            "pipeline_time_seconds": round(pipeline_time, 2)
+        }