feat(maritime): fix API rate limits, populate knowledge base, and upg…

backend/core/crew_maritime_compliance.py

@@ -41,6 +41,7 @@ def test_gemini_connection(api_key: str, timeout: int = 10) -> None:
 4) Confidence levels: HIGH (direct regulation match), MEDIUM (interpretation needed), LOW (unclear)
 5) Consider vessel type, flag state, and gross tonnage for applicability assessment
 6) Time-stamp findings and note any recent regulatory changes
+If specific long-form citations (e.g. Chapter X, Reg Y) are not available in search results, provide the most specific source and description found.
 Keep outputs concise and actionable.
 """
 
@@ -56,7 +57,7 @@ def _init_llm():
         # Test connection
         test_gemini_connection(api_key=settings.google_api_key)
         return LLM(
-            model="gemini/gemini-2.0-flash",
+            model="gemini/gemini-3-flash-preview",
             api_key=settings.google_api_key
         )
 
@@ -119,6 +120,8 @@ def build_maritime_compliance_crew(
         llm=llm,
         tools=tools,
         verbose=True,
+        max_iter=30,
+        max_rpm=3,
     )
 
     document_analyst = Agent(
@@ -128,6 +131,8 @@ def build_maritime_compliance_crew(
                   "document requirements, and identifying compliance gaps.",
         llm=llm,
         verbose=True,
+        max_iter=15,
+        max_rpm=3,
     )
 
     port_specialist = Agent(
@@ -138,6 +143,8 @@ def build_maritime_compliance_crew(
         llm=llm,
         tools=tools,
         verbose=True,
+        max_iter=20,
+        max_rpm=3,
     )
 
     risk_officer = Agent(
@@ -147,6 +154,8 @@ def build_maritime_compliance_crew(
                   "and operational impacts of compliance failures.",
         llm=llm,
         verbose=True,
+        max_iter=15,
+        max_rpm=3,
     )
 
     report_writer = Agent(
@@ -156,6 +165,7 @@ def build_maritime_compliance_crew(
                   "to ship operators and management.",
         llm=llm,
         verbose=True,
+        max_rpm=3,
     )
 
     # ========== TASKS ==========

backend/core/maritime_tools.py

@@ -27,13 +27,39 @@ def _run(self, *args, **kwargs):
 
 
 if HAS_CREWAI:
+    from pydantic import BaseModel, Field
     from services.maritime_knowledge_base import get_maritime_knowledge_base
     from services.compliance_service import ComplianceService
 
+    class MaritimeRAGSchema(BaseModel):
+        """Input for MaritimeRAGTool"""
+        query: str = Field(..., description="The search query for maritime regulations")
+        port_code: Optional[str] = Field(None, description="Optional port code filter")
+        region: Optional[str] = Field(None, description="Optional region filter")
+        regulation_type: Optional[str] = Field(None, description="Optional regulation type filter")
+        vessel_type: Optional[str] = Field(None, description="Optional vessel type filter")
+        top_k: int = Field(10, description="Number of results to return")
+
+    class PortInfoSchema(BaseModel):
+        """Input for PortInfoTool"""
+        port_code: str = Field(..., description="UN/LOCODE format port code")
+        vessel_type: Optional[str] = Field(None, description="Vessel type for filtering requirements")
+
+    class DocumentCheckSchema(BaseModel):
+        """Input for DocumentCheckTool"""
+        vessel_documents_json: Any = Field(..., description="JSON string or list/dict of vessel's documents")
+        required_documents_json: Any = Field(..., description="JSON string or list of required document types")
+
+    class RouteAnalysisSchema(BaseModel):
+        """Input for RouteAnalysisTool"""
+        route_ports_json: Any = Field(..., description="JSON array or list of port codes")
+        vessel_type: str = Field("container", description="Type of vessel")
+
     class MaritimeRAGTool(BaseTool):
         """Tool for querying maritime regulations knowledge base"""
 
         name: str = "maritime_regulations_search"
+        args_schema: Any = MaritimeRAGSchema
         description: str = """
         Search the maritime regulations knowledge base for laws, conventions, and requirements.
 
@@ -60,10 +86,20 @@ def _run(
             region: Optional[str] = None,
             regulation_type: Optional[str] = None,
             vessel_type: Optional[str] = None,
-            top_k: int = 5
+            top_k: int = 10,
+            **kwargs
         ) -> str:
             """Execute the search"""
             try:
+                # Handle CrewAI passing all args as a dict to the first parameter
+                if isinstance(query, dict) and "query" in query:
+                    port_code = query.get("port_code", port_code)
+                    region = query.get("region", region)
+                    regulation_type = query.get("regulation_type", regulation_type)
+                    vessel_type = query.get("vessel_type", vessel_type)
+                    top_k = query.get("top_k", top_k)
+                    query = query["query"]
+
                 kb = get_maritime_knowledge_base()
 
                 # Build filters
@@ -93,7 +129,7 @@ def _run(
                 for i, result in enumerate(results, 1):
                     source = result.metadata.get("source_convention", result.source)
                     output_lines.append(f"[{i}] Source: {source}")
-                    output_lines.append(f"    Content: {result.content[:300]}...")
+                    output_lines.append(f"    Content: {result.content[:1500]}...")
                     if result.metadata:
                         meta_str = ", ".join(f"{k}={v}" for k, v in result.metadata.items()
                                              if k not in ["source_convention"])
@@ -112,6 +148,7 @@ class PortInfoTool(BaseTool):
         """Tool for getting port information and requirements"""
 
         name: str = "port_info"
+        args_schema: Any = PortInfoSchema
         description: str = """
         Get detailed information about a specific port including:
         - Port State Control regime (Paris MOU, Tokyo MOU, etc.)
@@ -127,10 +164,16 @@ class PortInfoTool(BaseTool):
         def _run(
             self,
             port_code: str,
-            vessel_type: Optional[str] = None
+            vessel_type: Optional[str] = None,
+            **kwargs
         ) -> str:
             """Get port information"""
             try:
+                # Handle CrewAI passing all args as a dict to the first parameter
+                if isinstance(port_code, dict) and "port_code" in port_code:
+                    vessel_type = port_code.get("vessel_type", vessel_type)
+                    port_code = port_code["port_code"]
+
                 kb = get_maritime_knowledge_base()
 
                 # Search for port-specific info
@@ -154,7 +197,7 @@ def _run(
                 output_lines.append("\nAPPLICABLE REGULATIONS:")
                 if port_results:
                     for result in port_results[:5]:
-                        output_lines.append(f"- {result.content[:200]}...")
+                        output_lines.append(f"- {result.content[:1000]}...")
                         output_lines.append(f"  (Source: {result.metadata.get('source', result.source)})")
                 else:
                     output_lines.append("- No specific regulations found in database")
@@ -179,6 +222,7 @@ class DocumentCheckTool(BaseTool):
         """Tool for checking document compliance"""
 
         name: str = "document_check"
+        args_schema: Any = DocumentCheckSchema
         description: str = """
         Check what documents a vessel has and identify gaps.
 
@@ -190,13 +234,20 @@ class DocumentCheckTool(BaseTool):
 
         def _run(
             self,
-            vessel_documents_json: str,
-            required_documents_json: str
+            vessel_documents_json: Any,
+            required_documents_json: Any,
+            **kwargs
         ) -> str:
             """Check documents against requirements"""
             try:
-                vessel_docs = json.loads(vessel_documents_json)
-                required_docs = json.loads(required_documents_json)
+                # Handle CrewAI passing all args as a dict to the first parameter
+                if isinstance(vessel_documents_json, dict) and "vessel_documents_json" in vessel_documents_json:
+                    required_documents_json = vessel_documents_json.get("required_documents_json", required_documents_json)
+                    vessel_documents_json = vessel_documents_json["vessel_documents_json"]
+
+                # Handle input that might already be list/dict from newer CrewAI
+                vessel_docs = vessel_documents_json if isinstance(vessel_documents_json, (list, dict)) else json.loads(vessel_documents_json)
+                required_docs = required_documents_json if isinstance(required_documents_json, list) else json.loads(required_documents_json)
 
                 # Build document type set from vessel docs
                 available_types = set()
@@ -253,6 +304,7 @@ class RouteAnalysisTool(BaseTool):
         """Tool for analyzing route compliance"""
 
         name: str = "route_analysis"
+        args_schema: Any = RouteAnalysisSchema
         description: str = """
         Analyze a shipping route to identify all applicable regulations and requirements.
 
@@ -265,12 +317,19 @@ class RouteAnalysisTool(BaseTool):
 
         def _run(
             self,
-            route_ports_json: str,
-            vessel_type: str = "container"
+            route_ports_json: Any,
+            vessel_type: str = "container",
+            **kwargs
         ) -> str:
             """Analyze route compliance"""
             try:
-                port_codes = json.loads(route_ports_json)
+                # Handle CrewAI passing all args as a dict to the first parameter
+                if isinstance(route_ports_json, dict) and "route_ports_json" in route_ports_json:
+                    vessel_type = route_ports_json.get("vessel_type", vessel_type)
+                    route_ports_json = route_ports_json["route_ports_json"]
+
+                # Handle input that might already be a list
+                port_codes = route_ports_json if isinstance(route_ports_json, list) else json.loads(route_ports_json)
 
                 if not port_codes:
                     return "No ports provided in route"
@@ -294,7 +353,7 @@ def _run(
                     if port_results:
                         output_lines.append("Regulations:")
                         for result in port_results[:2]:
-                            output_lines.append(f"  - {result.content[:150]}...")
+                            output_lines.append(f"  - {result.content[:1000]}...")
 
                     if required_docs:
                         output_lines.append("Required Documents:")

backend/scripts/ingest_maritime_data.py

@@ -347,9 +347,6 @@ def ingest_imo_conventions(self):
 
         documents = create_documents_from_data(IMO_CONVENTIONS_DATA, "imo_conventions")
 
-        if self.kb.mock_mode:
-            logger.warning("Knowledge base in mock mode - documents not stored")
-            return len(documents)
 
         count = self.kb.add_documents("imo_conventions", documents)
         logger.info(f"Ingested {count} IMO convention documents")
@@ -361,9 +358,6 @@ def ingest_psc_requirements(self):
 
         documents = create_documents_from_data(PSC_REQUIREMENTS_DATA, "psc_requirements")
 
-        if self.kb.mock_mode:
-            logger.warning("Knowledge base in mock mode - documents not stored")
-            return len(documents)
 
         count = self.kb.add_documents("psc_requirements", documents)
         logger.info(f"Ingested {count} PSC requirement documents")
@@ -375,9 +369,6 @@ def ingest_regional_requirements(self):
 
         documents = create_documents_from_data(REGIONAL_REQUIREMENTS_DATA, "regional_requirements")
 
-        if self.kb.mock_mode:
-            logger.warning("Knowledge base in mock mode - documents not stored")
-            return len(documents)
 
         count = self.kb.add_documents("regional_requirements", documents)
         logger.info(f"Ingested {count} regional requirement documents")
@@ -395,9 +386,6 @@ def ingest_from_json_file(self, json_path: str, collection_name: str):
 
         documents = create_documents_from_data(data, collection_name)
 
-        if self.kb.mock_mode:
-            logger.warning("Knowledge base in mock mode - documents not stored")
-            return len(documents)
 
         count = self.kb.add_documents(collection_name, documents)
         logger.info(f"Ingested {count} documents from {json_path}")

backend/services/maritime_knowledge_base.py

@@ -55,20 +55,28 @@ def __init__(self):
 
         # Initialize Gemini embeddings
         self.embeddings = GoogleGenerativeAIEmbeddings(
-            model="gemini-embedding-001",
+            model="models/gemini-embedding-001",
             google_api_key=settings.google_api_key
         )
 
         # Create a Chroma collection for each defined collection
         for collection_name in self.COLLECTIONS.keys():
-            self.collections[collection_name] = Chroma(
-                collection_name=collection_name,
-                embedding_function=self.embeddings,
-                chroma_cloud_api_key=settings.chroma_api_key,
-                tenant=settings.chroma_tenant,
-                database=settings.chroma_database,
-            )
-            logger.info(f"Initialized collection: {collection_name}")
+            # Use local persistence if no cloud API key is provided
+            if not settings.chroma_api_key:
+                self.collections[collection_name] = Chroma(
+                    collection_name=collection_name,
+                    embedding_function=self.embeddings,
+                    persist_directory=settings.maritime_kb_persist_dir
+                )
+            else:
+                self.collections[collection_name] = Chroma(
+                    collection_name=collection_name,
+                    embedding_function=self.embeddings,
+                    chroma_cloud_api_key=settings.chroma_api_key,
+                    tenant=settings.chroma_tenant,
+                    database=settings.chroma_database,
+                )
+            logger.info(f"Initialized collection: {collection_name} (Local: {not settings.chroma_api_key})")
 
         # Initialize cross-encoder reranker
         self.reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
@@ -90,10 +98,18 @@ def search_by_port(
         """
         # Build query and filters
         query = f"Port requirements regulations for port {port_code}"
-        filters = {"port_code": port_code}
+
+        # ChromaDB requires $and for multiple filters
         if vessel_type:
-            filters["vessel_type"] = vessel_type
-
+            filters = {
+                "$and": [
+                    {"port_code": port_code},
+                    {"vessel_type": vessel_type}
+                ]
+            }
+        else:
+            filters = {"port_code": port_code}
+
         # Search across relevant collections
         results = []
         for collection_name in ["port_regulations", "psc_requirements", "customs_documentation"]: