Coverage for src / lexigram / contracts / ai / protocols.py: 0%

1"""Core AI provider and subsystem protocols. 

2 

3Defines the two top-level structural contracts for AI packages: 

4 

5* ``AIProviderProtocol`` — any package that acts as an AI provider (chat, etc.) 

6* ``AISubsystemProtocol`` — any AI subsystem package discovered via entry points 

7 

8For other AI contracts import from the focused sub-modules directly: 

9 

10* ``lexigram.contracts.ai.llm`` — LLM clients and prompt protocols 

11* ``lexigram.contracts.ai.vector`` — Vector store and document protocols 

12* ``lexigram.contracts.ai.rag`` — RAG pipeline protocols 

13* ``lexigram.contracts.ai.guards`` — GuardProtocol chain protocols 

14* ``lexigram.contracts.ai.agents`` — Agent strategy and memory protocols 

15""" 

16 

17from __future__ import annotations 

18 

19from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable 

20 

21from lexigram.contracts.core.provider import ProviderProtocol 

22 

23if TYPE_CHECKING: 

24 from lexigram.contracts.ai.llm import ChatMessageProtocol, CompletionProtocol 

25 

26 

27@runtime_checkable 

28class AIProviderProtocol(ProviderProtocol, Protocol): 

29 """Protocol for AI provider implementations.""" 

30 

31 async def chat( 

32 self, 

33 messages: list[ChatMessageProtocol], 

34 tools: list[dict[str, Any]] | None = None, 

35 **kwargs: Any, 

36 ) -> CompletionProtocol: 

37 """Chat with optional tool calling. 

38 

39 Args: 

40 messages: List of chat messages. 

41 tools: Optional list of JSON Schema tool definitions. 

42 Intentionally ``list[dict[str, Any]]`` — tool schemas are 

43 provider-specific JSON and their shape is not constrained here. 

44 **kwargs: Provider-specific options. 

45 

46 Returns: 

47 Completion with optional tool calls. 

48 """ 

49 ... 

50 

51 

52@runtime_checkable 

53class AISubsystemProtocol(Protocol): 

54 """Contract for AI subsystem packages discovered via entry points.""" 

55 

56 name: str 

57 

58 async def initialize(self) -> None: 

59 """Initialize the subsystem after all providers are booted.""" 

60 ... 

61 

62 async def health_check(self) -> bool: 

63 """Return True if the subsystem is healthy.""" 

64 ... 

65 

66 

67__all__ = [ 

68 "AIProviderProtocol", 

69 "AISubsystemProtocol", 

70]