Spaces:
Running
Running
File size: 4,433 Bytes
c2ea5ed |
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 |
"""
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 |