Spaces:
Running
Running
| """ | |
| Method Service for Knowledge Extraction Methods | |
| This service provides backend operations for managing and using knowledge extraction methods. | |
| """ | |
| from typing import Dict, Any, List, Optional | |
| from agentgraph.shared.method_registry import ( | |
| get_available_methods, | |
| get_method_info, | |
| get_method_names, | |
| get_production_methods, | |
| get_baseline_methods, | |
| get_method_display_name, | |
| get_method_description, | |
| get_schema_for_method, | |
| is_valid_method, | |
| DEFAULT_METHOD, | |
| MethodType, | |
| SchemaType | |
| ) | |
| from agentgraph.shared.extraction_factory import ( | |
| method_requires_content_references, | |
| method_requires_line_numbers, | |
| method_supports_failure_detection, | |
| get_method_processing_type | |
| ) | |
| class MethodService: | |
| """Service for managing knowledge extraction methods""" | |
| def get_available_methods(self) -> Dict[str, Any]: | |
| """Get all available methods with their metadata""" | |
| methods = get_available_methods() | |
| # Transform for API response | |
| result = {} | |
| for method_name, method_info in methods.items(): | |
| result[method_name] = { | |
| "name": method_info["name"], | |
| "description": method_info["description"], | |
| "method_type": method_info["method_type"].value, | |
| "schema_type": method_info["schema_type"].value, | |
| "supported_features": method_info["supported_features"], | |
| "processing_type": method_info["processing_type"] | |
| } | |
| return result | |
| def get_method_info(self, method_name: str) -> Optional[Dict[str, Any]]: | |
| """Get information about a specific method""" | |
| if not is_valid_method(method_name): | |
| return None | |
| method_info = get_method_info(method_name) | |
| return { | |
| "name": method_info["name"], | |
| "description": method_info["description"], | |
| "method_type": method_info["method_type"].value, | |
| "schema_type": method_info["schema_type"].value, | |
| "supported_features": method_info["supported_features"], | |
| "processing_type": method_info["processing_type"], | |
| "requires_content_references": method_requires_content_references(method_name), | |
| "requires_line_numbers": method_requires_line_numbers(method_name), | |
| "supports_failure_detection": method_supports_failure_detection(method_name) | |
| } | |
| # Removed separate production/baseline methods - use get_available_methods() with filtering | |
| def validate_method(self, method_name: str) -> Dict[str, Any]: | |
| """Validate a method name and return validation result""" | |
| if not method_name: | |
| return { | |
| "valid": False, | |
| "error": "Method name is required" | |
| } | |
| if not is_valid_method(method_name): | |
| available_methods = get_method_names() | |
| return { | |
| "valid": False, | |
| "error": f"Unknown method '{method_name}'. Available methods: {', '.join(available_methods)}" | |
| } | |
| return { | |
| "valid": True, | |
| "method_info": self.get_method_info(method_name) | |
| } | |
| def get_default_method(self) -> str: | |
| """Get the default method name""" | |
| return DEFAULT_METHOD | |
| def get_method_schema_compatibility(self, method_name: str) -> Dict[str, Any]: | |
| """Get schema compatibility information for a method""" | |
| if not is_valid_method(method_name): | |
| return {"error": f"Unknown method: {method_name}"} | |
| schema_type = get_schema_for_method(method_name) | |
| return { | |
| "method_name": method_name, | |
| "schema_type": schema_type.value, | |
| "requires_content_references": method_requires_content_references(method_name), | |
| "requires_line_numbers": method_requires_line_numbers(method_name), | |
| "supports_failure_detection": method_supports_failure_detection(method_name), | |
| "processing_type": get_method_processing_type(method_name) | |
| } | |
| # Removed separate filter methods - filtering is now done at the API level in get_available_methods() | |
| # Global service instance | |
| _method_service = MethodService() | |
| def get_method_service() -> MethodService: | |
| """Get the global method service instance""" | |
| return _method_service |