File size: 8,646 Bytes
b30f068 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 | #!/usr/bin/env python3
"""
Comprehensive system verification script for the Agentic AI System.
Checks all components, dependencies, and connections.
"""
import sys
import importlib
from pathlib import Path
from typing import List, Dict, Any
# Add src to path
sys.path.insert(0, str(Path(__file__).parent / "src"))
def check_component(name: str, module_path: str, class_name: str = None) -> Dict[str, Any]:
"""Check if a component can be imported and instantiated."""
try:
module = importlib.import_module(module_path)
if class_name:
cls = getattr(module, class_name)
# Try to instantiate (basic check)
instance = cls()
return {"status": "β
", "message": f"{name} working", "instance": instance}
else:
return {"status": "β
", "message": f"{name} module imported"}
except ImportError as e:
return {"status": "β", "message": f"{name} import failed: {e}"}
except Exception as e:
return {"status": "β οΈ", "message": f"{name} instantiation failed: {e}"}
def verify_system() -> None:
"""Perform comprehensive system verification."""
print("π Agentic AI System - Comprehensive Verification")
print("=" * 60)
# Core components to verify
components = [
# Configuration
("Settings", "config.settings", None),
# Parsers
("SpaCy Parser", "src.parsers.spacy_parser", "SpacyParser"),
("Keyword Extractor", "src.parsers.keyword_extractor", "KeywordExtractor"),
# API Clients
("API Matcher", "src.api_clients.api_matcher", "APIMatcher"),
("CDMS Client", "src.api_clients.cdms_client", "CDMSClient"),
# Agents
("Base Agent", "src.agents.base_agent", None),
("Supervisor Agent", "src.agents.supervisor_agent", "SupervisorAgent"),
("Query Parser Agent", "src.agents.worker_agents", "QueryParserAgent"),
("API Matcher Agent", "src.agents.worker_agents", "APIMatcherAgent"),
("API Executor Agent", "src.agents.worker_agents", "APIExecutorAgent"),
("Result Formatter Agent", "src.agents.worker_agents", "ResultFormatterAgent"),
# NEW: Evaluation Components
("Reflection Evaluator", "src.evaluators.reflection_evaluator", "ReflectionEvaluator"),
("Quality Assessor", "src.evaluators.quality_assessor", "QualityAssessor"),
("Memory Manager", "src.evaluators.memory_manager", "MemoryManager"),
# NEW: RAG System
("Hybrid RAG System", "src.rag.hybrid_rag", "HybridRAGSystem"),
# Utilities
("Logging Config", "src.utils.logging_config", None),
]
print("\\nπ¦ Component Verification:")
print("-" * 40)
results = []
for name, module_path, class_name in components:
result = check_component(name, module_path, class_name)
results.append((name, result))
print(f"{result['status']} {name}: {result['message']}")
# Check dependencies
print("\\nπ§ Dependency Verification:")
print("-" * 40)
dependencies = [
("FastAPI", "fastapi"),
("Uvicorn", "uvicorn"),
("spaCy", "spacy"),
("LangChain", "langchain"),
("LangGraph", "langgraph"),
("KeyBERT", "keybert"),
("RapidFuzz", "rapidfuzz"),
("Sentence Transformers", "sentence_transformers"),
("Qdrant Client", "qdrant_client"),
("NumPy", "numpy"),
("Pydantic", "pydantic"),
("Requests", "requests"),
]
dep_results = []
for name, module in dependencies:
try:
importlib.import_module(module)
print(f"β
{name}: Available")
dep_results.append((name, True))
except ImportError:
print(f"β {name}: Missing")
dep_results.append((name, False))
# Check spaCy model
print("\\nπ§ NLP Model Verification:")
print("-" * 40)
try:
import spacy
nlp = spacy.load("en_core_web_sm")
print("β
spaCy en_core_web_sm: Available")
spacy_model = True
except OSError:
print("β spaCy en_core_web_sm: Missing")
print(" Run: python -m spacy download en_core_web_sm")
spacy_model = False
# Check file structure
print("\\nπ File Structure Verification:")
print("-" * 40)
required_files = [
"api_server.py",
"run_api.py",
"test_api_simple.py",
"requirements.txt",
"README.md",
"config/settings.py",
"src/__init__.py",
"src/agents/__init__.py",
"src/parsers/__init__.py",
"src/api_clients/__init__.py",
"src/evaluators/__init__.py", # NEW
"src/rag/__init__.py", # NEW
"src/utils/__init__.py",
"tests/test_parsers.py",
"tests/evaluation_harness.py", # NEW
]
file_results = []
for file_path in required_files:
path = Path(file_path)
if path.exists():
print(f"β
{file_path}: Exists")
file_results.append((file_path, True))
else:
print(f"β {file_path}: Missing")
file_results.append((file_path, False))
# Test API server import
print("\\nπ API Server Verification:")
print("-" * 40)
try:
from api_server import app
print("β
API Server: Import successful")
api_server = True
except Exception as e:
print(f"β API Server: Import failed - {e}")
api_server = False
# Test evaluation harness
print("\\nπ§ͺ Evaluation Harness Verification:")
print("-" * 40)
try:
from tests.evaluation_harness import EvaluationHarness
print("β
Evaluation Harness: Import successful")
eval_harness = True
except Exception as e:
print(f"β Evaluation Harness: Import failed - {e}")
eval_harness = False
# Summary
print("\\n" + "=" * 60)
print("π VERIFICATION SUMMARY")
print("=" * 60)
# Component summary
component_success = len([r for _, r in results if r["status"] == "β
"])
component_total = len(results)
print(f"π¦ Components: {component_success}/{component_total} working")
# Dependency summary
dep_success = len([r for _, r in dep_results if r])
dep_total = len(dep_results)
print(f"π§ Dependencies: {dep_success}/{dep_total} available")
# File summary
file_success = len([r for _, r in file_results if r])
file_total = len(file_results)
print(f"π Files: {file_success}/{file_total} present")
# Overall status
overall_score = (component_success + dep_success + file_success) / (component_total + dep_total + file_total)
print(f"\\nπ― Overall System Health: {overall_score:.1%}")
if overall_score >= 0.9:
print("π SYSTEM READY TO GO! β¨")
print("\\nπ Quick Start:")
print(" 1. python run_api.py")
print(" 2. python test_api_simple.py")
print(" 3. python tests/evaluation_harness.py")
elif overall_score >= 0.7:
print("β οΈ SYSTEM MOSTLY READY - Minor issues detected")
print("\\nπ‘ Check missing components above")
else:
print("β SYSTEM NOT READY - Major issues detected")
print("\\nπ§ Install missing dependencies:")
print(" pip install -r requirements.txt")
if not spacy_model:
print(" python -m spacy download en_core_web_sm")
# Advanced features status
print("\\nπ Advanced Features Status:")
print("-" * 40)
advanced_features = [
("Self-Critique Reflection", "src.evaluators.reflection_evaluator"),
("Persistent Memory", "src.evaluators.memory_manager"),
("Quality Assessment", "src.evaluators.quality_assessor"),
("Hybrid RAG System", "src.rag.hybrid_rag"),
("Evaluation Harness", "tests.evaluation_harness"),
]
for name, module in advanced_features:
try:
importlib.import_module(module)
print(f"β
{name}: Available")
except ImportError:
print(f"β {name}: Not available")
print("\\n" + "=" * 60)
if __name__ == "__main__":
verify_system()
|