Agent Zero
76 lines
2.6 KiB
Python
76 lines
2.6 KiB
Python
"""Unified Search plugin contract — public interface for cross-plugin access."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from app.plugins.builtins.contracts import get_contract_registry
|
|
from app.plugins.builtins.unified_search.base_provider import BaseSearchProvider
|
|
from app.plugins.builtins.unified_search.embedding import generate_embedding
|
|
from app.plugins.builtins.unified_search.provider_registry import get_search_registry
|
|
from app.plugins.builtins.unified_search.query_understanding import llm_analyze_query
|
|
from app.plugins.builtins.unified_search.search_engine import find_similar_all_types, hybrid_search
|
|
|
|
|
|
async def simple_search(
|
|
db: Any,
|
|
query: str,
|
|
tenant_id: Any,
|
|
entity_types: list[str] | None = None,
|
|
limit: int = 20,
|
|
user_id: Any | None = None,
|
|
is_system_admin: bool = False,
|
|
) -> list[dict[str, Any]]:
|
|
"""Convenience search entry point: analyze a raw query string and run
|
|
the hybrid search over all registered providers.
|
|
|
|
Falls back to a plain normalized-query analysis when the LLM is
|
|
unavailable.
|
|
"""
|
|
analysis = await llm_analyze_query(query, db=db, tenant_id=tenant_id)
|
|
return await hybrid_search(
|
|
db=db,
|
|
query_analysis=analysis,
|
|
tenant_id=tenant_id,
|
|
entity_types=entity_types,
|
|
limit=limit,
|
|
user_id=user_id,
|
|
is_system_admin=is_system_admin,
|
|
)
|
|
|
|
|
|
class UnifiedSearchContract:
|
|
"""Public contract for the unified_search plugin."""
|
|
|
|
contract_name = "unified_search"
|
|
|
|
generate_embedding = staticmethod(generate_embedding)
|
|
hybrid_search = staticmethod(hybrid_search)
|
|
find_similar_all_types = staticmethod(find_similar_all_types)
|
|
get_search_registry = staticmethod(get_search_registry)
|
|
llm_analyze_query = staticmethod(llm_analyze_query)
|
|
simple_search = staticmethod(simple_search)
|
|
BaseSearchProvider = BaseSearchProvider
|
|
|
|
@classmethod
|
|
def get_function(cls, name: str):
|
|
"""Return a callable exposed by this contract, or None if absent."""
|
|
return getattr(cls, name, None)
|
|
|
|
|
|
# ─── self-registration ───
|
|
|
|
_contract = UnifiedSearchContract()
|
|
get_contract_registry().register("unified_search", _contract)
|
|
|
|
# Backward-compatible local accessor
|
|
_contract_instance: UnifiedSearchContract | None = None
|
|
|
|
|
|
def get_contract() -> UnifiedSearchContract:
|
|
global _contract_instance
|
|
if _contract_instance is None:
|
|
_contract_instance = UnifiedSearchContract()
|
|
return _contract_instance
|
|
|
|
|
|
__all__ = ["UnifiedSearchContract", "generate_embedding", "hybrid_search", "find_similar_all_types", "get_search_registry", "BaseSearchProvider"]
|