Coverage for astrocyte/integrations/camel_ai.py: 95%

1"""CAMEL-AI integration — Astrocyte memory for multi-agent role-playing. 

2 

3Usage: 

4 from astrocyte import Astrocyte 

5 from astrocyte.integrations.camel_ai import AstrocyteCamelMemory 

6 

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

8 memory = AstrocyteCamelMemory(brain, bank_id="simulation") 

9 

10 # Use with CAMEL agents 

11 memory.write("The patient reports headaches", role="doctor", agent_id="agent-1") 

12 context = await memory.read("patient symptoms", role="doctor") 

13 

14CAMEL-AI uses role-playing conversations between agents. Each agent has a role 

15and memories are scoped by role and/or agent identity. 

16""" 

17 

18from __future__ import annotations 

19 

20from typing import TYPE_CHECKING, Any 

21 

22if TYPE_CHECKING: 

23 from astrocyte._astrocyte import Astrocyte 

24 

25from astrocyte.types import AstrocyteContext 

26 

27 

28class AstrocyteCamelMemory: 

29 """Astrocyte-backed memory for CAMEL-AI multi-agent systems. 

30 

31 Supports role-based memory scoping: each role can have its own bank 

32 or share a common bank. Memory is tagged with role and agent_id for filtering. 

33 """ 

34 

35 def __init__( 

36 self, 

37 brain: Astrocyte, 

38 bank_id: str, 

39 *, 

40 context: AstrocyteContext | None = None, 

41 role_banks: dict[str, str] | None = None, 

42 ) -> None: 

43 self.brain = brain 

44 self.bank_id = bank_id 

45 self._context = context 

46 self._role_banks = role_banks or {} 

47 

48 def _resolve_bank(self, role: str | None = None) -> str: 

49 if role and role in self._role_banks: 

50 return self._role_banks[role] 

51 return self.bank_id 

52 

53 async def write( 

54 self, 

55 content: str, 

56 *, 

57 role: str | None = None, 

58 agent_id: str | None = None, 

59 tags: list[str] | None = None, 

60 ) -> str | None: 

61 """Write to memory. Returns memory_id.""" 

62 bank = self._resolve_bank(role) 

63 meta: dict[str, Any] = {"source": "camel-ai"} 

64 if role: 

65 meta["role"] = role 

66 if agent_id: 

67 meta["agent_id"] = agent_id 

68 

69 all_tags = list(tags or []) 

70 all_tags.append("camel-ai") 

71 if role: 

72 all_tags.append(f"role:{role}") 

73 

74 result = await self.brain.retain(content, bank_id=bank, tags=all_tags, metadata=meta, context=self._context) 

75 return result.memory_id if result.stored else None 

76 

77 async def read( 

78 self, 

79 query: str, 

80 *, 

81 role: str | None = None, 

82 max_results: int = 5, 

83 ) -> list[dict[str, Any]]: 

84 """Read from memory. Returns list of hit dicts.""" 

85 bank = self._resolve_bank(role) 

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

87 return [ 

88 {"text": h.text, "score": h.score, "metadata": h.metadata, "memory_id": h.memory_id} for h in result.hits 

89 ] 

90 

91 async def get_context(self, query: str, *, role: str | None = None, max_results: int = 5) -> str: 

92 """Get formatted memory context for injection into agent messages.""" 

93 hits = await self.read(query, role=role, max_results=max_results) 

94 if not hits: 

95 return "" 

96 return "\n".join(f"- {h['text']}" for h in hits) 

97 

98 async def reflect(self, query: str, *, role: str | None = None) -> str: 

99 """Synthesize an answer from memory.""" 

100 bank = self._resolve_bank(role) 

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

102 return result.answer 

103 

104 async def clear(self, *, role: str | None = None) -> None: 

105 """Clear all memory for a role or the shared bank.""" 

106 bank = self._resolve_bank(role) 

107 await self.brain.clear_bank(bank, context=self._context)