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

1"""Workflow protocols for the Lexigram Framework. 

2 

3Defines the structural contracts for the workflow graph execution 

4engine in ``lexigram-ai-workflow``. 

5""" 

6 

7from __future__ import annotations 

8 

9from dataclasses import dataclass, field 

10from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable 

11 

12from lexigram.contracts.ai.exceptions import WorkflowError 

13 

14if TYPE_CHECKING: 

15 from lexigram.contracts.core.result import Result 

16 

17 

18@dataclass(frozen=True) 

19class WorkflowResult: 

20 """Final result of a workflow graph execution. 

21 

22 Carries the terminal shared state, any structured output produced 

23 by the workflow, and an execution trace for debugging. 

24 """ 

25 

26 final_state: dict[str, Any] 

27 output: Any = None 

28 trace: list[str] = field(default_factory=list) 

29 

30 def get(self, key: str, default: Any = None) -> Any: 

31 """Retrieve a key from the final state with an optional default.""" 

32 return self.final_state.get(key, default) 

33 

34 

35@runtime_checkable 

36class AIWorkflowNodeProtocol(Protocol): 

37 """AI-domain graph node protocol for workflow execution. 

38 

39 A workflow node represents a single unit of work in a directed graph. 

40 Nodes receive the current shared state, perform their work, and 

41 return a state update dict that is merged into the global state. 

42 """ 

43 

44 @property 

45 def name(self) -> str: 

46 """Unique node identifier within the workflow.""" 

47 ... 

48 

49 async def execute(self, state: dict[str, Any]) -> dict[str, Any]: 

50 """Execute this node with the current workflow state. 

51 

52 Args: 

53 state: Current shared workflow state. 

54 

55 Returns: 

56 Dict of state updates to merge into the shared state. 

57 """ 

58 ... 

59 

60 

61@runtime_checkable 

62class WorkflowProtocol(Protocol): 

63 """Structural protocol for executable workflows. 

64 

65 A workflow is a directed graph of :class:`AIWorkflowNodeProtocol` nodes 

66 connected by conditional or unconditional edges. Execution is async, 

67 stateful, and supports cycles (with a max-iteration guard). 

68 """ 

69 

70 async def execute( 

71 self, 

72 input: str, 

73 *, 

74 config: Any | None = None, 

75 state: dict[str, Any] | None = None, 

76 ) -> Result[WorkflowResult, WorkflowError]: 

77 """Execute the workflow graph. 

78 

79 Args: 

80 input: Initial user input injected into the workflow state. 

81 config: Optional workflow configuration (e.g. max_iterations). 

82 state: Optional pre-populated initial state. 

83 

84 Returns: 

85 ``Ok(WorkflowResult)`` with the final state and execution trace, 

86 or ``Err(WorkflowError)`` on failure. 

87 """ 

88 ... 

89 

90 

91__all__ = [ 

92 "AIWorkflowNodeProtocol", 

93 "WorkflowProtocol", 

94 "WorkflowResult", 

95]