Coverage for astrocyte/integrations/semantic_kernel.py: 86%

1"""Microsoft Semantic Kernel integration — Astrocyte as kernel plugins. 

2 

3Usage: 

4 from astrocyte import Astrocyte 

5 from astrocyte.integrations.semantic_kernel import AstrocytePlugin 

6 

7 brain = Astrocyte.from_config("astrocyte.yaml") 

8 plugin = AstrocytePlugin(brain, bank_id="user-123") 

9 

10 # Register with Semantic Kernel 

11 kernel = Kernel() 

12 kernel.add_plugin(plugin, plugin_name="memory") 

13 

14Semantic Kernel uses a plugin/function pattern. Each plugin method decorated 

15with @kernel_function becomes a callable function. We emulate this pattern 

16with plain methods that follow the same signature conventions. 

17""" 

18 

19from __future__ import annotations 

20 

21from typing import TYPE_CHECKING, Any 

22 

23if TYPE_CHECKING: 

24 from astrocyte._astrocyte import Astrocyte 

25 

26from astrocyte.types import AstrocyteContext 

27 

28 

29class AstrocytePlugin: 

30 """Astrocyte memory plugin for Microsoft Semantic Kernel. 

31 

32 Methods follow the Semantic Kernel @kernel_function pattern: 

33 - Named methods with docstrings (used as function descriptions) 

34 - Type-annotated parameters (used for schema generation) 

35 - Return strings (Semantic Kernel standard for text results) 

36 

37 Register with: kernel.add_plugin(plugin, plugin_name="memory") 

38 """ 

39 

40 def __init__( 

41 self, 

42 brain: Astrocyte, 

43 bank_id: str, 

44 *, 

45 context: AstrocyteContext | None = None, 

46 include_reflect: bool = True, 

47 include_forget: bool = False, 

48 ) -> None: 

49 self.brain = brain 

50 self.bank_id = bank_id 

51 self._context = context 

52 self._include_reflect = include_reflect 

53 self._include_forget = include_forget 

54 

55 async def retain(self, content: str, tags: str = "") -> str: 

56 """Store content into long-term memory for future recall. 

57 

58 Args: 

59 content: The text to memorize. 

60 tags: Comma-separated tags for filtering (optional). 

61 """ 

62 tag_list = [t.strip() for t in tags.split(",") if t.strip()] if tags else None 

63 result = await self.brain.retain(content, bank_id=self.bank_id, tags=tag_list, context=self._context) 

64 if result.stored: 

65 return f"Stored memory (id: {result.memory_id})" 

66 return f"Failed to store: {result.error}" 

67 

68 async def recall(self, query: str, max_results: int = 5) -> str: 

69 """Search long-term memory for information relevant to a query. 

70 

71 Args: 

72 query: Natural language search query. 

73 max_results: Maximum number of results to return. 

74 """ 

75 result = await self.brain.recall(query, bank_id=self.bank_id, max_results=max_results, context=self._context) 

76 if not result.hits: 

77 return "No relevant memories found." 

78 lines = [f"- [{h.score:.2f}] {h.text}" for h in result.hits] 

79 return f"Found {len(result.hits)} memories:\n" + "\n".join(lines) 

80 

81 async def reflect(self, query: str) -> str: 

82 """Synthesize a comprehensive answer from long-term memory. 

83 

84 Args: 

85 query: The question to answer from memory. 

86 """ 

87 result = await self.brain.reflect(query, bank_id=self.bank_id, context=self._context) 

88 return result.answer 

89 

90 async def forget(self, memory_ids: str) -> str: 

91 """Remove specific memories by their IDs. 

92 

93 Args: 

94 memory_ids: Comma-separated memory IDs to delete. 

95 """ 

96 ids = [mid.strip() for mid in memory_ids.split(",")] 

97 result = await self.brain.forget(self.bank_id, memory_ids=ids, context=self._context) 

98 return f"Deleted {result.deleted_count} memories." 

99 

100 def get_functions(self) -> list[dict[str, Any]]: 

101 """Return the list of available functions for registration. 

102 

103 Each dict has 'name', 'description', 'function' for Semantic Kernel. 

104 """ 

105 fns: list[dict[str, Any]] = [ 

106 {"name": "retain", "description": self.retain.__doc__ or "", "function": self.retain}, 

107 {"name": "recall", "description": self.recall.__doc__ or "", "function": self.recall}, 

108 ] 

109 if self._include_reflect: 

110 fns.append({"name": "reflect", "description": self.reflect.__doc__ or "", "function": self.reflect}) 

111 if self._include_forget: 

112 fns.append({"name": "forget", "description": self.forget.__doc__ or "", "function": self.forget}) 

113 return fns