diff --git a/backend/core/crew_maritime_compliance.py b/backend/core/crew_maritime_compliance.py index 282876cf4..06375eda6 100644 --- a/backend/core/crew_maritime_compliance.py +++ b/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 ========== diff --git a/backend/core/maritime_tools.py b/backend/core/maritime_tools.py index 48dad151d..0b3e61434 100644 --- a/backend/core/maritime_tools.py +++ b/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:") diff --git a/backend/scripts/ingest_maritime_data.py b/backend/scripts/ingest_maritime_data.py index d052796f3..747fbce28 100644 --- a/backend/scripts/ingest_maritime_data.py +++ b/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}") diff --git a/backend/services/maritime_knowledge_base.py b/backend/services/maritime_knowledge_base.py index e0476f1f7..7e77ae873 100644 --- a/backend/services/maritime_knowledge_base.py +++ b/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"]: