86 lines
3.0 KiB
Python
86 lines
3.0 KiB
Python
from typing import Dict, Type, Any, Optional, Callable
|
|
from config import settings
|
|
from file_processor import TextFileProcessor
|
|
from vector_store import VectorStore
|
|
from hybrid_retriever import HybridRetriever
|
|
from generator import TextGenerator
|
|
from exporter import Exporter
|
|
from logger import GenerationLogger
|
|
from content_analyzer import analyzer
|
|
|
|
class DIContainer:
|
|
"""依赖注入容器"""
|
|
|
|
def __init__(self):
|
|
self._services: Dict[str, Any] = {}
|
|
self._singletons: Dict[str, Any] = {}
|
|
|
|
def register(self, name: str, service_class: Type, is_singleton: bool = False):
|
|
"""注册服务"""
|
|
self._services[name] = {
|
|
'class': service_class,
|
|
'singleton': is_singleton
|
|
}
|
|
|
|
def register_factory(self, name: str, factory: Callable, is_singleton: bool = False):
|
|
"""注册工厂函数"""
|
|
self._services[name] = {
|
|
'factory': factory,
|
|
'singleton': is_singleton
|
|
}
|
|
|
|
def register_instance(self, name: str, instance: Any, is_singleton: bool = True):
|
|
"""注册已创建的实例"""
|
|
if is_singleton:
|
|
self._singletons[name] = instance
|
|
self._services[name] = {
|
|
'instance': instance,
|
|
'singleton': is_singleton
|
|
}
|
|
|
|
def get(self, name: str, **kwargs) -> Any:
|
|
"""获取服务实例"""
|
|
if name not in self._services:
|
|
raise ValueError(f"Service {name} not registered")
|
|
|
|
service_info = self._services[name]
|
|
|
|
# 如果是单例，先检查是否已创建
|
|
if service_info['singleton']:
|
|
if name in self._singletons:
|
|
return self._singletons[name]
|
|
|
|
# 创建实例
|
|
if 'class' in service_info:
|
|
# 检查是否需要传递参数
|
|
if kwargs:
|
|
instance = service_info['class'](**kwargs)
|
|
else:
|
|
instance = service_info['class']()
|
|
elif 'factory' in service_info:
|
|
instance = service_info['factory']()
|
|
elif 'instance' in service_info:
|
|
instance = service_info['instance']
|
|
else:
|
|
raise ValueError(f"Service {name} has no valid creation method")
|
|
|
|
if service_info['singleton']:
|
|
self._singletons[name] = instance
|
|
|
|
return instance
|
|
|
|
# 创建全局容器实例
|
|
container = DIContainer()
|
|
|
|
# 注册所有服务
|
|
container.register('file_processor', TextFileProcessor, is_singleton=True)
|
|
container.register('vector_store', VectorStore, is_singleton=True)
|
|
container.register('text_generator', TextGenerator, is_singleton=True)
|
|
container.register('exporter', Exporter, is_singleton=True)
|
|
container.register('logger', GenerationLogger, is_singleton=True)
|
|
container.register_factory('analyzer', lambda: analyzer, is_singleton=True)
|
|
|
|
# 注册需要参数的服务
|
|
container.register_factory('hybrid_retriever',
|
|
lambda: HybridRetriever(container.get('vector_store')),
|
|
is_singleton=True) |