You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ch...@apache.org on 2026/03/01 14:58:54 UTC
(carbondata) branch master updated: add quickstart and demo for Agent module
This is an automated email from the ASF dual-hosted git repository.
chenliang613 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/carbondata.git
The following commit(s) were added to refs/heads/master by this push:
new 3c61921442 add quickstart and demo for Agent module
3c61921442 is described below
commit 3c61921442fd5a3acc660dc222667496dea83424
Author: chenliang613 <ch...@apache.org>
AuthorDate: Sun Mar 1 22:58:36 2026 +0800
add quickstart and demo for Agent module
---
Agent_module/Example/01_data_models.py | 482 ++++++++++++++++++++++++++++
Agent_module/Example/02_framework_agents.py | 249 ++++++++++++++
Agent_module/Example/03_agent_manager.py | 271 ++++++++++++++++
Agent_module/Example/04_context_store.py | 389 ++++++++++++++++++++++
Agent_module/Example/_loader.py | 49 +++
Agent_module/Example/run_all.py | 52 +++
Agent_module/QUICKSTART.md | 414 ++++++++++++++++++++++++
7 files changed, 1906 insertions(+)
diff --git a/Agent_module/Example/01_data_models.py b/Agent_module/Example/01_data_models.py
new file mode 100644
index 0000000000..9d6b1e0ef3
--- /dev/null
+++ b/Agent_module/Example/01_data_models.py
@@ -0,0 +1,482 @@
+"""
+01_data_models.py
+=================
+Tests every data model class used by the LLM-oriented Agent format:
+ ToolDefinition, ToolCall, Message, Memory, Task, AgentState, AgentConfig, Agent
+
+Run:
+ python3 Agent_module/Example/01_data_models.py
+"""
+
+import sys
+import os
+import json
+from datetime import datetime
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Literal, Optional
+from uuid import uuid4
+
+# ============================================================================
+# Data Models (self-contained)
+# ============================================================================
+
+@dataclass
+class ToolDefinition:
+ name: str
+ description: str
+ parameters: Dict[str, Any]
+ required: List[str] = field(default_factory=list)
+
+ def to_dict(self) -> Dict[str, Any]:
+ return {
+ "type": "function",
+ "function": {
+ "name": self.name,
+ "description": self.description,
+ "parameters": {
+ "type": "object",
+ "properties": self.parameters,
+ "required": self.required,
+ },
+ },
+ }
+
+
+@dataclass
+class ToolCall:
+ id: str = field(default_factory=lambda: f"call_{uuid4().hex[:16]}")
+ name: str = ""
+ arguments: Dict[str, Any] = field(default_factory=dict)
+ result: Optional[Any] = None
+ status: Literal["pending", "success", "failed"] = "pending"
+ error: Optional[str] = None
+ timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
+
+ def to_dict(self) -> Dict[str, Any]:
+ return {
+ "id": self.id,
+ "name": self.name,
+ "arguments": self.arguments,
+ "result": self.result,
+ "status": self.status,
+ "error": self.error,
+ "timestamp": self.timestamp,
+ }
+
+
+@dataclass
+class Message:
+ role: Literal["system", "user", "assistant", "tool"]
+ content: str
+ id: str = field(default_factory=lambda: f"msg_{uuid4().hex[:16]}")
+ name: Optional[str] = None
+ tool_calls: List[ToolCall] = field(default_factory=list)
+ tool_call_id: Optional[str] = None
+ metadata: Dict[str, Any] = field(default_factory=dict)
+ timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
+
+ def to_dict(self) -> Dict[str, Any]:
+ data: Dict[str, Any] = {
+ "id": self.id,
+ "role": self.role,
+ "content": self.content,
+ "timestamp": self.timestamp,
+ }
+ if self.name:
+ data["name"] = self.name
+ if self.tool_calls:
+ data["tool_calls"] = [tc.to_dict() for tc in self.tool_calls]
+ if self.tool_call_id:
+ data["tool_call_id"] = self.tool_call_id
+ if self.metadata:
+ data["metadata"] = self.metadata
+ return data
+
+
+@dataclass
+class Memory:
+ short_term: List[Message] = field(default_factory=list)
+ long_term: List[Dict[str, Any]] = field(default_factory=list)
+ working_memory: Dict[str, Any] = field(default_factory=dict)
+ max_short_term: int = 20
+
+ def add_to_short_term(self, message: Message) -> None:
+ self.short_term.append(message)
+ if len(self.short_term) > self.max_short_term:
+ self.short_term.pop(0)
+
+ def add_to_long_term(self, key: str, value: Any, importance: float = 0.5) -> None:
+ self.long_term.append({
+ "key": key,
+ "value": value,
+ "importance": importance,
+ "timestamp": datetime.now().isoformat(),
+ })
+
+ def to_dict(self) -> Dict[str, Any]:
+ return {
+ "short_term": [m.to_dict() for m in self.short_term],
+ "long_term": self.long_term,
+ "working_memory": self.working_memory,
+ }
+
+
+@dataclass
+class Task:
+ id: str = field(default_factory=lambda: f"task_{uuid4().hex[:16]}")
+ description: str = ""
+ status: Literal["pending", "in_progress", "completed", "failed"] = "pending"
+ priority: int = 0
+ subtasks: List["Task"] = field(default_factory=list)
+ result: Optional[Any] = None
+ created_at: str = field(default_factory=lambda: datetime.now().isoformat())
+ completed_at: Optional[str] = None
+
+ def to_dict(self) -> Dict[str, Any]:
+ return {
+ "id": self.id,
+ "description": self.description,
+ "status": self.status,
+ "priority": self.priority,
+ "subtasks": [st.to_dict() for st in self.subtasks],
+ "result": self.result,
+ "created_at": self.created_at,
+ "completed_at": self.completed_at,
+ }
+
+
+@dataclass
+class AgentState:
+ mode: Literal["idle", "thinking", "acting", "waiting"] = "idle"
+ current_task: Optional[Task] = None
+ task_queue: List[Task] = field(default_factory=list)
+ context: Dict[str, Any] = field(default_factory=dict)
+ iteration_count: int = 0
+ max_iterations: int = 10
+ last_action: Optional[str] = None
+ last_updated: str = field(default_factory=lambda: datetime.now().isoformat())
+
+ def to_dict(self) -> Dict[str, Any]:
+ return {
+ "mode": self.mode,
+ "current_task": self.current_task.to_dict() if self.current_task else None,
+ "task_queue": [t.to_dict() for t in self.task_queue],
+ "context": self.context,
+ "iteration_count": self.iteration_count,
+ "max_iterations": self.max_iterations,
+ "last_action": self.last_action,
+ "last_updated": self.last_updated,
+ }
+
+
+@dataclass
+class AgentConfig:
+ model: str = "claude-sonnet-4-5"
+ temperature: float = 0.7
+ max_tokens: int = 4096
+ top_p: float = 0.9
+ enable_memory: bool = True
+ enable_reasoning: bool = True
+ reasoning_effort: Literal["low", "medium", "high"] = "medium"
+ streaming: bool = False
+
+ def to_dict(self) -> Dict[str, Any]:
+ return {
+ "model": self.model,
+ "temperature": self.temperature,
+ "max_tokens": self.max_tokens,
+ "top_p": self.top_p,
+ "enable_memory": self.enable_memory,
+ "enable_reasoning": self.enable_reasoning,
+ "reasoning_effort": self.reasoning_effort,
+ "streaming": self.streaming,
+ }
+
+
+@dataclass
+class Agent:
+ id: str = field(default_factory=lambda: f"agent_{uuid4().hex[:12]}")
+ name: str = "Assistant"
+ role: str = "general_assistant"
+ description: str = "A helpful AI assistant"
+ system_prompt: str = "You are a helpful AI assistant."
+ config: AgentConfig = field(default_factory=AgentConfig)
+ tools: List[ToolDefinition] = field(default_factory=list)
+ memory: Memory = field(default_factory=Memory)
+ state: AgentState = field(default_factory=AgentState)
+ created_at: str = field(default_factory=lambda: datetime.now().isoformat())
+ version: str = "1.0.0"
+ metadata: Dict[str, Any] = field(default_factory=dict)
+
+ def add_tool(self, tool: ToolDefinition) -> None:
+ self.tools.append(tool)
+
+ def add_message(self, message: Message) -> None:
+ self.memory.add_to_short_term(message)
+
+ def create_task(self, description: str, priority: int = 0) -> Task:
+ task = Task(description=description, priority=priority)
+ self.state.task_queue.append(task)
+ self.state.task_queue.sort(key=lambda t: t.priority, reverse=True)
+ return task
+
+ def update_state(self, **kwargs: Any) -> None:
+ for key, value in kwargs.items():
+ if hasattr(self.state, key):
+ setattr(self.state, key, value)
+ self.state.last_updated = datetime.now().isoformat()
+
+ def to_dict(self) -> Dict[str, Any]:
+ return {
+ "id": self.id,
+ "name": self.name,
+ "role": self.role,
+ "description": self.description,
+ "system_prompt": self.system_prompt,
+ "config": self.config.to_dict(),
+ "tools": [t.to_dict() for t in self.tools],
+ "memory": self.memory.to_dict(),
+ "state": self.state.to_dict(),
+ "created_at": self.created_at,
+ "version": self.version,
+ "metadata": self.metadata,
+ }
+
+ def to_json(self, indent: int = 2) -> str:
+ return json.dumps(self.to_dict(), indent=indent, ensure_ascii=False)
+
+
+# ============================================================================
+# Test helpers
+# ============================================================================
+
+_passed = 0
+_failed = 0
+
+
+def check(label: str, condition: bool) -> None:
+ global _passed, _failed
+ if condition:
+ _passed += 1
+ print(f" PASS {label}")
+ else:
+ _failed += 1
+ print(f" FAIL {label}")
+
+
+def section(title: str) -> None:
+ print(f"\n{'=' * 60}")
+ print(f" {title}")
+ print(f"{'=' * 60}")
+
+
+# ============================================================================
+# Tests
+# ============================================================================
+
+def test_tool_definition() -> None:
+ section("ToolDefinition")
+ tool = ToolDefinition(
+ name="web_search",
+ description="Search the web",
+ parameters={"query": {"type": "string", "description": "search term"}},
+ required=["query"],
+ )
+ d = tool.to_dict()
+ check("type is 'function'", d["type"] == "function")
+ check("name preserved", d["function"]["name"] == "web_search")
+ check("required list preserved", d["function"]["parameters"]["required"] == ["query"])
+ check("query parameter present", "query" in d["function"]["parameters"]["properties"])
+
+
+def test_tool_call() -> None:
+ section("ToolCall")
+ tc = ToolCall(name="execute_code", arguments={"code": "print(1)"}, result="1", status="success")
+ d = tc.to_dict()
+ check("id auto-generated", tc.id.startswith("call_"))
+ check("name preserved", d["name"] == "execute_code")
+ check("status is success", d["status"] == "success")
+ check("result preserved", d["result"] == "1")
+ check("error is None by default", d["error"] is None)
+
+ failed_tc = ToolCall(name="bad_tool", status="failed", error="timeout")
+ check("error field set", failed_tc.to_dict()["error"] == "timeout")
+
+
+def test_message() -> None:
+ section("Message")
+ tc = ToolCall(name="search", arguments={"q": "hello"}, status="success", result="found")
+
+ user_msg = Message(role="user", content="Hello", metadata={"user_id": "u1"})
+ d = user_msg.to_dict()
+ check("id auto-generated", user_msg.id.startswith("msg_"))
+ check("role preserved", d["role"] == "user")
+ check("content preserved", d["content"] == "Hello")
+ check("metadata included when set", "metadata" in d)
+ check("tool_calls absent when empty", "tool_calls" not in d)
+
+ asst_msg = Message(role="assistant", content="I'll search for that", tool_calls=[tc])
+ d2 = asst_msg.to_dict()
+ check("tool_calls included when set", "tool_calls" in d2)
+ check("tool_call name correct", d2["tool_calls"][0]["name"] == "search")
+
+ tool_msg = Message(role="tool", content="found", name="search", tool_call_id=tc.id)
+ d3 = tool_msg.to_dict()
+ check("tool_call_id preserved", d3["tool_call_id"] == tc.id)
+ check("name preserved", d3["name"] == "search")
+
+
+def test_memory() -> None:
+ section("Memory")
+ mem = Memory(max_short_term=3)
+
+ for i in range(5):
+ mem.add_to_short_term(Message(role="user", content=f"msg {i}"))
+
+ check("sliding window enforced (max 3)", len(mem.short_term) == 3)
+ check("oldest messages dropped", mem.short_term[0].content == "msg 2")
+ check("newest message retained", mem.short_term[-1].content == "msg 4")
+
+ mem.add_to_long_term("fact_1", "sky is blue", importance=0.9)
+ check("long-term entry added", len(mem.long_term) == 1)
+ check("importance stored", mem.long_term[0]["importance"] == 0.9)
+
+ mem.working_memory["current_topic"] = "testing"
+ d = mem.to_dict()
+ check("working_memory serialised", d["working_memory"]["current_topic"] == "testing")
+ check("short_term serialised", len(d["short_term"]) == 3)
+ check("long_term serialised", len(d["long_term"]) == 1)
+
+
+def test_task() -> None:
+ section("Task")
+ parent = Task(description="Write tests", priority=10)
+ child = Task(description="Write unit tests", priority=5)
+ parent.subtasks.append(child)
+
+ check("id auto-generated", parent.id.startswith("task_"))
+ check("default status is pending", parent.status == "pending")
+ check("subtask appended", len(parent.subtasks) == 1)
+
+ parent.status = "completed"
+ parent.result = "All tests written"
+ parent.completed_at = datetime.now().isoformat()
+ d = parent.to_dict()
+ check("status updated", d["status"] == "completed")
+ check("result preserved", d["result"] == "All tests written")
+ check("completed_at set", d["completed_at"] is not None)
+ check("subtask serialised", len(d["subtasks"]) == 1)
+
+
+def test_agent_state() -> None:
+ section("AgentState")
+ state = AgentState()
+ check("default mode is idle", state.mode == "idle")
+ check("task_queue starts empty", len(state.task_queue) == 0)
+
+ task = Task(description="do something")
+ state.current_task = task
+ state.mode = "acting"
+ state.iteration_count = 2
+ d = state.to_dict()
+ check("mode updated", d["mode"] == "acting")
+ check("iteration_count updated", d["iteration_count"] == 2)
+ check("current_task serialised", d["current_task"]["description"] == "do something")
+
+
+def test_agent_config() -> None:
+ section("AgentConfig")
+ cfg = AgentConfig(model="claude-opus-4", temperature=0.2, reasoning_effort="high")
+ d = cfg.to_dict()
+ check("model preserved", d["model"] == "claude-opus-4")
+ check("temperature preserved", d["temperature"] == 0.2)
+ check("reasoning_effort preserved", d["reasoning_effort"] == "high")
+ check("defaults preserved", d["max_tokens"] == 4096)
+
+
+def test_agent_full() -> None:
+ section("Agent — full lifecycle")
+
+ agent = Agent(
+ name="CodeBot",
+ role="developer",
+ description="Writes code",
+ system_prompt="You write clean Python.",
+ config=AgentConfig(model="claude-sonnet-4-5", temperature=0.3),
+ )
+
+ # Tools
+ agent.add_tool(ToolDefinition(
+ name="run_code",
+ description="Execute Python",
+ parameters={"code": {"type": "string"}},
+ required=["code"],
+ ))
+ check("tool registered", len(agent.tools) == 1)
+
+ # Messages
+ agent.add_message(Message(role="system", content=agent.system_prompt))
+ agent.add_message(Message(role="user", content="Write fibonacci"))
+ agent.add_message(Message(role="assistant", content="Here it is"))
+ check("messages in short-term memory", len(agent.memory.short_term) == 3)
+
+ # Tasks — priority ordering
+ t_low = agent.create_task("low priority task", priority=1)
+ t_high = agent.create_task("high priority task", priority=9)
+ check("high-priority task sorted first", agent.state.task_queue[0].priority == 9)
+ check("low-priority task sorted second", agent.state.task_queue[1].priority == 1)
+
+ # State update
+ agent.update_state(mode="thinking", current_task=t_high, iteration_count=1)
+ check("mode updated via update_state", agent.state.mode == "thinking")
+ check("current_task set", agent.state.current_task.description == "high priority task")
+
+ # Serialisation round-trip
+ d = agent.to_dict()
+ check("id starts with agent_", d["id"].startswith("agent_"))
+ check("name preserved", d["name"] == "CodeBot")
+ check("config.model preserved", d["config"]["model"] == "claude-sonnet-4-5")
+ check("tools serialised", len(d["tools"]) == 1)
+ check("memory.short_term serialised", len(d["memory"]["short_term"]) == 3)
+ check("state.mode serialised", d["state"]["mode"] == "thinking")
+
+ # to_json / valid JSON
+ json_str = agent.to_json()
+ parsed = json.loads(json_str)
+ check("to_json produces valid JSON", parsed["name"] == "CodeBot")
+
+
+# ============================================================================
+# Entry point
+# ============================================================================
+
+def run() -> bool:
+ global _passed, _failed
+ _passed = _failed = 0
+
+ print("\n" + "=" * 60)
+ print(" 01 — Agent Data Models")
+ print("=" * 60)
+
+ test_tool_definition()
+ test_tool_call()
+ test_message()
+ test_memory()
+ test_task()
+ test_agent_state()
+ test_agent_config()
+ test_agent_full()
+
+ total = _passed + _failed
+ print(f"\n{'=' * 60}")
+ print(f" Result: {_passed}/{total} passed", "✓" if _failed == 0 else "✗")
+ print(f"{'=' * 60}")
+ return _failed == 0
+
+
+if __name__ == "__main__":
+ ok = run()
+ sys.exit(0 if ok else 1)
diff --git a/Agent_module/Example/02_framework_agents.py b/Agent_module/Example/02_framework_agents.py
new file mode 100644
index 0000000000..a2deb930b4
--- /dev/null
+++ b/Agent_module/Example/02_framework_agents.py
@@ -0,0 +1,249 @@
+"""
+02_framework_agents.py
+======================
+Tests BaseAgent subclassing, AgentStatus lifecycle, message history,
+callbacks, and error handling from Framework.py and Implementation.py.
+
+Run:
+ python3 Agent_module/Example/02_framework_agents.py
+"""
+
+import asyncio
+import sys
+import os
+import time
+
+sys.path.insert(0, os.path.dirname(__file__))
+
+from _loader import load
+
+_m = load("Framework", "Implementation")
+BaseAgent = _m.BaseAgent
+AgentMessage = _m.AgentMessage
+AgentStatus = _m.AgentStatus
+EchoAgent = _m.EchoAgent
+CalculatorAgent = _m.CalculatorAgent
+WeatherAgent = _m.WeatherAgent
+TranslatorAgent = _m.TranslatorAgent
+
+# ============================================================================
+# Test helpers
+# ============================================================================
+
+_passed = 0
+_failed = 0
+
+
+def check(label: str, condition: bool) -> None:
+ global _passed, _failed
+ if condition:
+ _passed += 1
+ print(f" PASS {label}")
+ else:
+ _failed += 1
+ print(f" FAIL {label}")
+
+
+def section(title: str) -> None:
+ print(f"\n{'=' * 60}")
+ print(f" {title}")
+ print(f"{'=' * 60}")
+
+
+def msg(content: str, **meta) -> AgentMessage:
+ return AgentMessage(content=content, sender="tester", timestamp=time.time(), metadata=meta)
+
+
+# ============================================================================
+# Custom agent used in tests
+# ============================================================================
+
+class UpperCaseAgent(BaseAgent):
+ """Returns the message content uppercased."""
+
+ def __init__(self):
+ super().__init__(name="UpperCaseAgent", description="Converts text to uppercase")
+
+ async def _process_impl(self, message: AgentMessage) -> AgentMessage:
+ return AgentMessage(
+ content=str(message.content).upper(),
+ sender=self.name,
+ timestamp=time.time(),
+ message_type="uppercase_response",
+ )
+
+
+class BrokenAgent(BaseAgent):
+ """Always raises an exception — tests error handling."""
+
+ def __init__(self):
+ super().__init__(name="BrokenAgent", description="Always fails")
+
+ async def _process_impl(self, message: AgentMessage) -> AgentMessage:
+ raise RuntimeError("intentional failure")
+
+
+# ============================================================================
+# Tests
+# ============================================================================
+
+async def test_custom_agent() -> None:
+ section("Custom agent — UpperCaseAgent")
+
+ agent = UpperCaseAgent()
+ check("initial status is IDLE", agent.status == AgentStatus.IDLE)
+ check("name set correctly", agent.name == "UpperCaseAgent")
+ check("history starts empty", len(agent.message_history) == 0)
+
+ response = await agent.process(msg("hello world"))
+ check("response content uppercased", response.content == "HELLO WORLD")
+ check("response sender is agent name", response.sender == "UpperCaseAgent")
+ check("message_type preserved", response.message_type == "uppercase_response")
+ check("status returns to IDLE", agent.status == AgentStatus.IDLE)
+ check("input added to history", len(agent.message_history) == 1)
+
+ # Send a second message
+ await agent.process(msg("second message"))
+ check("history grows with each message", len(agent.message_history) == 2)
+ check("get_recent_messages(1) returns 1", len(agent.get_recent_messages(1)) == 1)
+ check("get_recent_messages(5) returns 2", len(agent.get_recent_messages(5)) == 2)
+
+
+async def test_error_handling() -> None:
+ section("Error handling — BrokenAgent")
+
+ agent = BrokenAgent()
+ response = await agent.process(msg("trigger error"))
+
+ check("response message_type is error", response.message_type == "error")
+ check("response content contains 'Error'", "Error" in response.content)
+ check("status set to ERROR", agent.status == AgentStatus.ERROR)
+
+
+async def test_callbacks() -> None:
+ section("Callbacks")
+
+ agent = UpperCaseAgent()
+ log: list = []
+
+ def record(input_msg: AgentMessage, output_msg: AgentMessage) -> None:
+ log.append((input_msg.content, output_msg.content))
+
+ agent.add_callback(record)
+ await agent.process(msg("ping"))
+ await agent.process(msg("pong"))
+
+ check("callback fired for each message", len(log) == 2)
+ check("callback received correct input", log[0][0] == "ping")
+ check("callback received correct output", log[0][1] == "PING")
+ check("second callback entry correct", log[1] == ("pong", "PONG"))
+
+ # A crashing callback must not break the agent
+ def bad_callback(i, o):
+ raise ValueError("bad callback")
+
+ agent.add_callback(bad_callback)
+ response = await agent.process(msg("still works"))
+ check("agent survives a crashing callback", response.content == "STILL WORKS")
+
+
+async def test_echo_agent() -> None:
+ section("EchoAgent")
+
+ agent = EchoAgent()
+ response = await agent.process(msg("hello"))
+ check("response starts with 'Echo:'", response.content.startswith("Echo:"))
+ check("original content echoed", "hello" in response.content)
+ check("message_type is echo_response", response.message_type == "echo_response")
+
+
+async def test_calculator_agent() -> None:
+ section("CalculatorAgent")
+
+ agent = CalculatorAgent()
+
+ cases = [
+ ("2 + 2", "4"),
+ ("10 * 5", "50"),
+ ("100 / 4", "25.0"),
+ ("(3 + 7) * 2", "20"),
+ ]
+ for expr, expected in cases:
+ r = await agent.process(msg(expr))
+ check(f" {expr} = {expected}", expected in r.content)
+
+ # Invalid input should return an error message, not crash
+ r = await agent.process(msg("__import__('os')"))
+ check("invalid chars -> error message", "error" in r.content.lower() or "Error" in r.content)
+
+ r = await agent.process(msg("1 / 0"))
+ check("division by zero -> error message", "error" in r.content.lower() or "Error" in r.content)
+
+
+async def test_weather_agent() -> None:
+ section("WeatherAgent")
+
+ agent = WeatherAgent()
+
+ known_cities = ["New York", "London", "Tokyo", "Sydney"]
+ for city in known_cities:
+ r = await agent.process(msg(city))
+ check(f" {city}: contains city name", city in r.content)
+ check(f" {city}: contains Temperature", "Temperature" in r.content)
+ check(f" {city}: message_type correct", r.message_type == "weather_report")
+
+ # Unknown city should still return a response (simulated)
+ r = await agent.process(msg("Atlantis"))
+ check("unknown city returns simulated data", "Atlantis" in r.content)
+ check("simulated data has Temperature field", "Temperature" in r.content)
+
+
+async def test_translator_agent() -> None:
+ section("TranslatorAgent")
+
+ agent = TranslatorAgent()
+
+ supported = ["spanish", "french", "german", "italian"]
+ for lang in supported:
+ r = await agent.process(msg("Hello", target_language=lang))
+ check(f" {lang}: lang name in response", lang.upper() in r.content)
+ check(f" {lang}: message_type correct", r.message_type == "translation")
+
+ # Unsupported language
+ r = await agent.process(msg("Hello", target_language="klingon"))
+ check("unsupported language returns error text", "don't support" in r.content or "klingon" in r.content.lower())
+
+
+# ============================================================================
+# Entry point
+# ============================================================================
+
+async def _run_all() -> bool:
+ global _passed, _failed
+ _passed = _failed = 0
+
+ await test_custom_agent()
+ await test_error_handling()
+ await test_callbacks()
+ await test_echo_agent()
+ await test_calculator_agent()
+ await test_weather_agent()
+ await test_translator_agent()
+
+ total = _passed + _failed
+ print(f"\n{'=' * 60}")
+ print(f" Result: {_passed}/{total} passed", "✓" if _failed == 0 else "✗")
+ print(f"{'=' * 60}")
+ return _failed == 0
+
+
+def run() -> bool:
+ print("\n" + "=" * 60)
+ print(" 02 — Framework Agents")
+ print("=" * 60)
+ return asyncio.run(_run_all())
+
+
+if __name__ == "__main__":
+ ok = run()
+ sys.exit(0 if ok else 1)
diff --git a/Agent_module/Example/03_agent_manager.py b/Agent_module/Example/03_agent_manager.py
new file mode 100644
index 0000000000..92ea5a636a
--- /dev/null
+++ b/Agent_module/Example/03_agent_manager.py
@@ -0,0 +1,271 @@
+"""
+03_agent_manager.py
+===================
+Tests AgentManager: registration, unicast, broadcast, capability
+discovery, system statistics, and concurrent execution.
+
+Run:
+ python3 Agent_module/Example/03_agent_manager.py
+"""
+
+import asyncio
+import sys
+import os
+import time
+
+sys.path.insert(0, os.path.dirname(__file__))
+
+from _loader import load
+
+_m = load("Framework", "Implementation", "Manager", "Chat_Agent")
+BaseAgent = _m.BaseAgent
+AgentMessage = _m.AgentMessage
+AgentStatus = _m.AgentStatus
+AgentManager = _m.AgentManager
+EchoAgent = _m.EchoAgent
+CalculatorAgent = _m.CalculatorAgent
+WeatherAgent = _m.WeatherAgent
+TranslatorAgent = _m.TranslatorAgent
+ChatAgent = _m.ChatAgent
+
+# ============================================================================
+# Test helpers
+# ============================================================================
+
+_passed = 0
+_failed = 0
+
+
+def check(label: str, condition: bool) -> None:
+ global _passed, _failed
+ if condition:
+ _passed += 1
+ print(f" PASS {label}")
+ else:
+ _failed += 1
+ print(f" FAIL {label}")
+
+
+def section(title: str) -> None:
+ print(f"\n{'=' * 60}")
+ print(f" {title}")
+ print(f"{'=' * 60}")
+
+
+# ============================================================================
+# Helper agent
+# ============================================================================
+
+class TagAgent(BaseAgent):
+ """Returns a fixed tag — useful for verifying broadcast routing."""
+
+ def __init__(self, name: str, tag: str):
+ super().__init__(name=name, description=f"Returns tag: {tag}")
+ self.tag = tag
+
+ async def _process_impl(self, message: AgentMessage) -> AgentMessage:
+ return AgentMessage(
+ content=self.tag,
+ sender=self.name,
+ timestamp=time.time(),
+ )
+
+
+# ============================================================================
+# Tests
+# ============================================================================
+
+async def test_registration() -> None:
+ section("Agent registration")
+
+ manager = AgentManager()
+ a1 = EchoAgent()
+ a2 = CalculatorAgent()
+
+ manager.register_agent(a1)
+ manager.register_agent(a2)
+
+ check("two agents registered", len(manager.agents) == 2)
+ check("EchoAgent found by name", "EchoAgent" in manager.agents)
+ check("CalculatorAgent found by name", "CalculatorAgent" in manager.agents)
+
+ # Re-registering with the same name replaces the entry
+ a3 = EchoAgent()
+ manager.register_agent(a3)
+ check("re-register keeps count at 2", len(manager.agents) == 2)
+ check("replaced agent is the new one", manager.agents["EchoAgent"] is a3)
+
+
+async def test_send_message() -> None:
+ section("Unicast — send_message")
+
+ manager = AgentManager()
+ manager.register_agent(CalculatorAgent())
+ manager.register_agent(WeatherAgent())
+
+ r = await manager.send_message("CalculatorAgent", "6 * 7")
+ check("calculator returns correct result", "42" in r.content)
+ check("sender is CalculatorAgent", r.sender == "CalculatorAgent")
+
+ r = await manager.send_message("WeatherAgent", "Tokyo")
+ check("weather response contains Tokyo", "Tokyo" in r.content)
+
+ # Unknown agent should raise ValueError
+ try:
+ await manager.send_message("NonExistentAgent", "hello")
+ check("ValueError raised for unknown agent", False)
+ except ValueError:
+ check("ValueError raised for unknown agent", True)
+
+
+async def test_broadcast() -> None:
+ section("Broadcast — broadcast_message")
+
+ manager = AgentManager()
+ manager.register_agent(TagAgent("Alpha", "alpha-reply"))
+ manager.register_agent(TagAgent("Beta", "beta-reply"))
+ manager.register_agent(TagAgent("Gamma", "gamma-reply"))
+
+ responses = await manager.broadcast_message("ping")
+
+ check("all three agents responded", len(responses) == 3)
+ check("Alpha response correct", responses["Alpha"].content == "alpha-reply")
+ check("Beta response correct", responses["Beta"].content == "beta-reply")
+ check("Gamma response correct", responses["Gamma"].content == "gamma-reply")
+
+
+async def test_broadcast_with_failure() -> None:
+ section("Broadcast — partial failure handled gracefully")
+
+ class FailingAgent(BaseAgent):
+ def __init__(self):
+ super().__init__(name="FailingAgent", description="always fails")
+
+ async def _process_impl(self, m: AgentMessage) -> AgentMessage:
+ raise RuntimeError("boom")
+
+ manager = AgentManager()
+ manager.register_agent(TagAgent("Good", "ok"))
+ manager.register_agent(FailingAgent())
+
+ responses = await manager.broadcast_message("test")
+
+ check("both agents have a response entry", len(responses) == 2)
+ check("Good agent responded normally", responses["Good"].content == "ok")
+ # FailingAgent's error is caught inside BaseAgent.process(), not in broadcast
+ check("FailingAgent returned an error message", responses["FailingAgent"].message_type == "error")
+
+
+async def test_capability_discovery() -> None:
+ section("Capability discovery — find_agent_by_capability")
+
+ manager = AgentManager()
+ manager.register_agent(CalculatorAgent()) # description: "Performs mathematical calculations"
+ manager.register_agent(WeatherAgent()) # description: "Provides weather information"
+ manager.register_agent(TranslatorAgent()) # description: "Translates text between languages"
+
+ calc_agents = manager.find_agent_by_capability("calculation")
+ weather_agents = manager.find_agent_by_capability("weather")
+ trans_agents = manager.find_agent_by_capability("translat")
+ none_agents = manager.find_agent_by_capability("nonexistent_xyz")
+
+ check("calculator found by 'calculation'", "CalculatorAgent" in calc_agents)
+ check("weather found by 'weather'", "WeatherAgent" in weather_agents)
+ check("translator found by 'translat'", "TranslatorAgent" in trans_agents)
+ check("empty list for no match", none_agents == [])
+
+
+async def test_system_stats() -> None:
+ section("System statistics")
+
+ manager = AgentManager()
+ manager.register_agent(EchoAgent())
+ manager.register_agent(CalculatorAgent())
+
+ await manager.send_message("EchoAgent", "msg 1")
+ await manager.send_message("EchoAgent", "msg 2")
+ await manager.send_message("CalculatorAgent", "1 + 1")
+
+ stats = manager.get_system_stats()
+
+ check("total_agents is 2", stats["total_agents"] == 2)
+ check("total_messages_processed is 3", stats["total_messages_processed"] == 3)
+ check("conversation_log_entries is 3", stats["conversation_log_entries"] == 3)
+ check("agent_statuses dict present", "agent_statuses" in stats)
+ check("EchoAgent status is idle", stats["agent_statuses"]["EchoAgent"] == "idle")
+
+
+async def test_metadata_routing() -> None:
+ section("Metadata routing — TranslatorAgent target_language")
+
+ manager = AgentManager()
+ manager.register_agent(TranslatorAgent())
+
+ r = await manager.send_message(
+ "TranslatorAgent",
+ "Good morning",
+ metadata={"target_language": "german"}
+ )
+ check("german translation triggered", "GERMAN" in r.content)
+ check("original text present", "Good morning" in r.content)
+
+
+async def test_chat_agent() -> None:
+ section("ChatAgent — basic intent detection")
+
+ manager = AgentManager()
+ manager.register_agent(ChatAgent())
+
+ greetings = ["hello", "hi there", "hey", "greetings"]
+ for g in greetings:
+ r = await manager.send_message("ChatAgent", g, metadata={"user_id": "u1"})
+ check(f" greeting '{g}' gets a reply", len(r.content) > 0)
+ check(f" message_type is chat_response", r.message_type == "chat_response")
+
+ r = await manager.send_message("ChatAgent", "what is your name?", metadata={"user_id": "u1"})
+ check("question gets an answer", len(r.content) > 0)
+
+ # Conversation count should increment in context
+ r1 = await manager.send_message("ChatAgent", "hello", metadata={"user_id": "ctx_user"})
+ # Read count1 immediately — the metadata dict is a live reference that will
+ # be mutated by subsequent messages for the same user_id.
+ count1 = r1.metadata.get("user_context", {}).get("conversation_count", 0)
+ r2 = await manager.send_message("ChatAgent", "hello", metadata={"user_id": "ctx_user"})
+ count2 = r2.metadata.get("user_context", {}).get("conversation_count", 0)
+ check("conversation_count increments per user", count2 == count1 + 1)
+
+
+# ============================================================================
+# Entry point
+# ============================================================================
+
+async def _run_all() -> bool:
+ global _passed, _failed
+ _passed = _failed = 0
+
+ await test_registration()
+ await test_send_message()
+ await test_broadcast()
+ await test_broadcast_with_failure()
+ await test_capability_discovery()
+ await test_system_stats()
+ await test_metadata_routing()
+ await test_chat_agent()
+
+ total = _passed + _failed
+ print(f"\n{'=' * 60}")
+ print(f" Result: {_passed}/{total} passed", "✓" if _failed == 0 else "✗")
+ print(f"{'=' * 60}")
+ return _failed == 0
+
+
+def run() -> bool:
+ print("\n" + "=" * 60)
+ print(" 03 — AgentManager")
+ print("=" * 60)
+ return asyncio.run(_run_all())
+
+
+if __name__ == "__main__":
+ ok = run()
+ sys.exit(0 if ok else 1)
diff --git a/Agent_module/Example/04_context_store.py b/Agent_module/Example/04_context_store.py
new file mode 100644
index 0000000000..6f5f4650bd
--- /dev/null
+++ b/Agent_module/Example/04_context_store.py
@@ -0,0 +1,389 @@
+"""
+04_context_store.py
+===================
+Tests every major feature of ContextStore and ContextNamespace:
+ put / get / delete / history / keys / query
+ P2P channels (send / receive / inbox / peek)
+ Pub/Sub (subscribe / unsubscribe / callback ordering)
+ TTL expiry (lazy + batch cleanup)
+ Namespace isolation
+ Merge (last-write-wins by version)
+ Snapshot and to_json serialisation
+ Stats
+
+Run:
+ python3 Agent_module/Example/04_context_store.py
+"""
+
+import asyncio
+import sys
+import os
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from ContextStore import ContextStore, ContextNamespace, ContextRecord
+
+# ============================================================================
+# Test helpers
+# ============================================================================
+
+_passed = 0
+_failed = 0
+
+
+def check(label: str, condition: bool) -> None:
+ global _passed, _failed
+ if condition:
+ _passed += 1
+ print(f" PASS {label}")
+ else:
+ _failed += 1
+ print(f" FAIL {label}")
+
+
+def section(title: str) -> None:
+ print(f"\n{'=' * 60}")
+ print(f" {title}")
+ print(f"{'=' * 60}")
+
+
+# ============================================================================
+# Tests
+# ============================================================================
+
+async def test_basic_put_get() -> None:
+ section("Basic put / get / delete")
+
+ store = ContextStore()
+
+ rec = await store.put("color", "red", writer="alice")
+ check("put returns a ContextRecord", isinstance(rec, ContextRecord))
+ check("record key is correct", rec.key == "color")
+ check("record value is correct", rec.value == "red")
+ check("record writer is correct", rec.writer == "alice")
+ check("first write gets version 1", rec.version == 1)
+ check("record_id is a non-empty string", len(rec.record_id) > 0)
+
+ val = await store.get("color")
+ check("get returns current value", val == "red")
+
+ # Overwrite
+ rec2 = await store.put("color", "blue", writer="bob")
+ check("second write gets version 2", rec2.version == 2)
+ check("get returns updated value", await store.get("color") == "blue")
+
+ # Missing key
+ check("get missing key returns None", await store.get("__missing__") is None)
+
+ # Delete
+ deleted = await store.delete("color", writer="alice")
+ check("delete returns True for existing key", deleted)
+ check("get after delete returns None", await store.get("color") is None)
+
+ not_deleted = await store.delete("__missing__", writer="alice")
+ check("delete returns False for missing key", not not_deleted)
+
+
+async def test_history() -> None:
+ section("History")
+
+ store = ContextStore()
+
+ await store.put("status", "pending", writer="coordinator")
+ await store.put("status", "in_progress", writer="worker")
+ await store.put("status", "completed", writer="worker")
+
+ history = await store.history("status")
+ check("history returns 3 records", len(history) == 3)
+ check("history is newest-first", history[0].value == "completed")
+ check("oldest record is pending", history[-1].value == "pending")
+
+ limited = await store.history("status", limit=2)
+ check("limit=2 returns 2 records", len(limited) == 2)
+ check("limit respects newest-first order", limited[0].value == "completed")
+
+ empty = await store.history("__missing__")
+ check("history for missing key is []", empty == [])
+
+
+async def test_keys_and_query() -> None:
+ section("keys() and query()")
+
+ store = ContextStore()
+ ns = "test_ns"
+
+ await store.put("a", 1, writer="alice", namespace=ns, tags=["x"])
+ await store.put("b", 2, writer="bob", namespace=ns, tags=["x", "y"])
+ await store.put("c", 3, writer="alice", namespace=ns, tags=["y"])
+
+ all_keys = await store.keys(namespace=ns)
+ check("keys() returns all 3 keys", set(all_keys) == {"a", "b", "c"})
+
+ alice_entries = await store.query(namespace=ns, writer="alice")
+ check("query(writer='alice') returns 2 entries", len(alice_entries) == 2)
+ check("all alice entries have writer='alice'",
+ all(e.current.writer == "alice" for e in alice_entries))
+
+ tag_x = await store.query(namespace=ns, tags=["x"])
+ check("query(tags=['x']) returns 2 entries", len(tag_x) == 2)
+
+ tag_xy = await store.query(namespace=ns, tags=["x", "y"])
+ check("query(tags=['x','y']) returns 1 entry", len(tag_xy) == 1)
+ check("that entry is 'b'", tag_xy[0].current.key == "b")
+
+ no_match = await store.query(namespace=ns, writer="nobody")
+ check("query with no match returns []", no_match == [])
+
+
+async def test_namespace_isolation() -> None:
+ section("Namespace isolation")
+
+ store = ContextStore()
+
+ await store.put("key", "ns1-value", writer="agent", namespace="ns1")
+ await store.put("key", "ns2-value", writer="agent", namespace="ns2")
+
+ check("ns1 reads its own value", await store.get("key", namespace="ns1") == "ns1-value")
+ check("ns2 reads its own value", await store.get("key", namespace="ns2") == "ns2-value")
+ check("default ns is empty", await store.get("key") is None)
+
+ ns1_keys = await store.keys(namespace="ns1")
+ ns2_keys = await store.keys(namespace="ns2")
+ check("ns1 has exactly 1 key", ns1_keys == ["key"])
+ check("ns2 has exactly 1 key", ns2_keys == ["key"])
+
+
+async def test_context_namespace_proxy() -> None:
+ section("ContextNamespace proxy")
+
+ store = ContextStore()
+ ns = ContextNamespace(store, "proxy_ns")
+
+ check("namespace property correct", ns.namespace == "proxy_ns")
+
+ await ns.put("x", 42, writer="agent")
+ check("ns.put writes to correct namespace", await ns.get("x") == 42)
+ check("direct store read matches", await store.get("x", "proxy_ns") == 42)
+
+ await ns.delete("x", writer="agent")
+ check("ns.delete removes the key", await ns.get("x") is None)
+
+ await ns.put("h", "v1", writer="a")
+ await ns.put("h", "v2", writer="b")
+ hist = await ns.history("h")
+ check("ns.history returns 2 records", len(hist) == 2)
+
+ keys = await ns.keys()
+ check("ns.keys() returns ['h']", keys == ["h"])
+
+
+async def test_channels() -> None:
+ section("P2P Channels — send / receive / peek / inbox")
+
+ store = ContextStore()
+
+ # Send two messages from coordinator to worker
+ m1 = await store.send("coordinator", "worker", {"cmd": "run"})
+ m2 = await store.send("coordinator", "worker", {"cmd": "stop"})
+ check("send returns a ContextMessage", m1.sender == "coordinator")
+ check("message_id is unique", m1.message_id != m2.message_id)
+
+ ch = store.channel("coordinator", "worker")
+ check("channel depth is 2", len(ch) == 2)
+
+ # Peek does not consume
+ peeked = await ch.peek()
+ check("peek returns first message", peeked.payload == {"cmd": "run"})
+ check("peek does not consume (depth still 2)", len(ch) == 2)
+
+ # Receive consumes one
+ r = await store.receive("coordinator", "worker")
+ check("receive returns oldest message (FIFO)", r.payload == {"cmd": "run"})
+ check("channel depth is now 1", len(ch) == 1)
+
+ # Drain with inbox across multiple senders
+ await store.send("auditor", "worker", {"cmd": "check"})
+ inbox = await store.inbox("worker")
+ check("inbox collects from all senders", len(inbox) == 2)
+ check("inbox sorted oldest-first by timestamp", inbox[0].payload == {"cmd": "stop"})
+ check("inbox is destructive (channel empty)", len(store.channel("coordinator", "worker")) == 0)
+
+ # Empty channel
+ empty = await store.receive("coordinator", "worker")
+ check("receive on empty channel returns None", empty is None)
+
+ # Tags on messages
+ tagged = await store.send("a", "b", "payload", tags=["urgent"])
+ check("message tags preserved", "urgent" in tagged.tags)
+
+
+async def test_pubsub() -> None:
+ section("Pub/Sub")
+
+ store = ContextStore()
+ ns = ContextNamespace(store, "pubsub_ns")
+ log: list = []
+
+ async def async_cb(key: str, record: ContextRecord) -> None:
+ log.append(("async", key, record.value, record.writer))
+
+ def sync_cb(key: str, record: ContextRecord) -> None:
+ log.append(("sync", key, record.value, record.writer))
+
+ # Subscribe monitor (async) and logger (sync) to all keys
+ ns.subscribe("monitor", async_cb)
+ ns.subscribe("logger", sync_cb)
+
+ await ns.put("status", "started", writer="coordinator")
+ check("async callback fired", any(e[0] == "async" for e in log))
+ check("sync callback fired", any(e[0] == "sync" for e in log))
+ check("both callbacks fired (2 entries)", len(log) == 2)
+ check("correct key delivered", all(e[1] == "status" for e in log))
+ check("correct value delivered", all(e[2] == "started" for e in log))
+
+ # Writer does NOT receive their own notification
+ log.clear()
+ ns.subscribe("coordinator", lambda k, r: log.append("self-notified"))
+ await ns.put("status", "done", writer="coordinator")
+ check("writer not notified of own write", "self-notified" not in log)
+ check("other subscribers still notified", len(log) == 2)
+
+ # Key-filtered subscription
+ log.clear()
+ ns.unsubscribe("monitor")
+ ns.unsubscribe("logger")
+ ns.unsubscribe("coordinator")
+
+ ns.subscribe("watcher", lambda k, r: log.append(k), keys=["important"])
+ await ns.put("important", "yes", writer="a")
+ await ns.put("unimportant", "skip", writer="a")
+ check("key filter: watched key triggers callback", "important" in log)
+ check("key filter: unwatched key does not trigger", "unimportant" not in log)
+
+ # Unsubscribe
+ log.clear()
+ ns.unsubscribe("watcher")
+ await ns.put("important", "again", writer="a")
+ check("after unsubscribe, no callback fires", log == [])
+
+
+async def test_ttl() -> None:
+ section("TTL expiry")
+
+ store = ContextStore()
+ ns = ContextNamespace(store, "ttl_ns")
+
+ await ns.put("short", "alive", writer="agent", ttl=1)
+ check("immediate read returns value", await ns.get("short") == "alive")
+
+ await asyncio.sleep(1.2)
+ check("read after TTL returns None", await ns.get("short") is None)
+
+ # Batch cleanup
+ await ns.put("a", 1, writer="x", ttl=1)
+ await ns.put("b", 2, writer="x", ttl=1)
+ await ns.put("c", 3, writer="x") # no TTL — should survive
+
+ await asyncio.sleep(1.2)
+ removed = await ns.cleanup_expired()
+ check("cleanup_expired removed 2 keys", removed == 2)
+ check("non-expiring key 'c' still readable", await ns.get("c") == 3)
+
+
+async def test_merge() -> None:
+ section("Merge — last-write-wins by version")
+
+ store_a = ContextStore()
+ store_b = ContextStore()
+ ns = "merge_ns"
+
+ # store_a writes status twice (version 2)
+ await store_a.put("status", "v1", writer="a", namespace=ns)
+ await store_a.put("status", "v2", writer="a", namespace=ns)
+
+ # store_b writes status once (version 1) — lower version
+ await store_b.put("status", "b-v1", writer="b", namespace=ns)
+ await store_b.put("extra", "only-in-b", writer="b", namespace=ns)
+
+ # Merge b into a — b's status (v1) should NOT overwrite a's status (v2)
+ merged = await store_a.merge(store_b, namespace=ns)
+ check("merge returns count of accepted entries", merged >= 1)
+ check("higher-version local value retained", await store_a.get("status", ns) == "v2")
+ check("new key from b accepted", await store_a.get("extra", ns) == "only-in-b")
+
+ # Merge a into b — a's status (v2) should overwrite b's status (v1)
+ await store_b.merge(store_a, namespace=ns)
+ check("higher-version remote value wins on merge", await store_b.get("status", ns) == "v2")
+
+
+async def test_snapshot_and_serialisation() -> None:
+ section("Snapshot and serialisation")
+
+ import json
+
+ store = ContextStore()
+ ns = ContextNamespace(store, "serial_ns")
+
+ await ns.put("k1", "v1", writer="a", tags=["t1"])
+ await ns.put("k2", {"nested": True}, writer="b")
+ await store.send("a", "b", {"cmd": "go"})
+
+ # Snapshot
+ snap = await ns.snapshot()
+ check("snapshot has k1 and k2", set(snap.keys()) == {"k1", "k2"})
+ check("snapshot is a deep copy", snap["k1"]["current"]["value"] == "v1")
+
+ # to_json
+ json_str = await store.to_json()
+ parsed = json.loads(json_str)
+
+ check("to_json produces valid JSON",
+ "entries" in parsed and "channels" in parsed and "subscriptions" in parsed)
+ check("namespace present in JSON", "serial_ns" in parsed["entries"])
+ check("k1 present in JSON entries", "k1" in parsed["entries"]["serial_ns"])
+ check("channel captured in JSON", "a->b" in parsed["channels"])
+
+ # Stats
+ stats = await store.stats()
+ check("stats has namespaces key", "namespaces" in stats)
+ check("stats has total_channels key", "total_channels" in stats)
+ check("serial_ns appears in stats", "serial_ns" in stats["namespaces"])
+ check("total_channels >= 1", stats["total_channels"] >= 1)
+
+
+# ============================================================================
+# Entry point
+# ============================================================================
+
+async def _run_all() -> bool:
+ global _passed, _failed
+ _passed = _failed = 0
+
+ await test_basic_put_get()
+ await test_history()
+ await test_keys_and_query()
+ await test_namespace_isolation()
+ await test_context_namespace_proxy()
+ await test_channels()
+ await test_pubsub()
+ await test_ttl()
+ await test_merge()
+ await test_snapshot_and_serialisation()
+
+ total = _passed + _failed
+ print(f"\n{'=' * 60}")
+ print(f" Result: {_passed}/{total} passed", "✓" if _failed == 0 else "✗")
+ print(f"{'=' * 60}")
+ return _failed == 0
+
+
+def run() -> bool:
+ print("\n" + "=" * 60)
+ print(" 04 — ContextStore")
+ print("=" * 60)
+ return asyncio.run(_run_all())
+
+
+if __name__ == "__main__":
+ ok = run()
+ sys.exit(0 if ok else 1)
diff --git a/Agent_module/Example/_loader.py b/Agent_module/Example/_loader.py
new file mode 100644
index 0000000000..2e307b4bf4
--- /dev/null
+++ b/Agent_module/Example/_loader.py
@@ -0,0 +1,49 @@
+"""
+_loader.py
+==========
+Utility for loading Agent_module source files that share a namespace.
+
+Framework.py, Implementation.py, Manager.py, and Chat_Agent.py were written
+as a single-namespace codebase (no per-file imports). This loader execs them
+together into one shared dict so the cross-file dependencies resolve correctly.
+
+Usage:
+ from _loader import load
+ m = load("Framework", "Implementation", "Manager", "Chat_Agent")
+ BaseAgent = m.BaseAgent
+ EchoAgent = m.EchoAgent
+ AgentManager = m.AgentManager
+"""
+
+import os
+import types
+import asyncio
+import time
+import random
+import json
+
+_MODULE_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..")
+
+
+def load(*names: str) -> types.SimpleNamespace:
+ """
+ Execute each named .py file in _MODULE_DIR into a shared namespace
+ and return a SimpleNamespace with all defined names as attributes.
+
+ Example:
+ m = load("Framework", "Implementation")
+ agent = m.EchoAgent()
+ """
+ ns: dict = {
+ "__builtins__": __builtins__,
+ "asyncio": asyncio,
+ "time": time,
+ "random": random,
+ "json": json,
+ }
+ for name in names:
+ path = os.path.join(_MODULE_DIR, f"{name}.py")
+ with open(path) as fh:
+ exec(compile(fh.read(), path, "exec"), ns)
+
+ return types.SimpleNamespace(**{k: v for k, v in ns.items() if not k.startswith("_")})
diff --git a/Agent_module/Example/run_all.py b/Agent_module/Example/run_all.py
new file mode 100644
index 0000000000..4dcb120810
--- /dev/null
+++ b/Agent_module/Example/run_all.py
@@ -0,0 +1,52 @@
+"""
+run_all.py
+==========
+Runs all example test files and prints a summary.
+
+Run:
+ python3 Agent_module/Example/run_all.py
+"""
+
+import sys
+import os
+import importlib
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+sys.path.insert(0, os.path.dirname(__file__))
+
+EXAMPLES = [
+ ("01_data_models", "Agent Data Models"),
+ ("02_framework_agents", "Framework Agents"),
+ ("03_agent_manager", "AgentManager"),
+ ("04_context_store", "ContextStore"),
+]
+
+
+def main() -> None:
+ results: list = []
+
+ for module_name, label in EXAMPLES:
+ mod = importlib.import_module(module_name)
+ ok = mod.run()
+ results.append((label, ok))
+
+ # Summary
+ print("\n" + "=" * 60)
+ print(" SUMMARY")
+ print("=" * 60)
+ all_passed = True
+ for label, ok in results:
+ status = "PASS" if ok else "FAIL"
+ print(f" {status} {label}")
+ if not ok:
+ all_passed = False
+
+ print("=" * 60)
+ print(f" {'All examples passed ✓' if all_passed else 'Some examples FAILED ✗'}")
+ print("=" * 60)
+
+ sys.exit(0 if all_passed else 1)
+
+
+if __name__ == "__main__":
+ main()
diff --git a/Agent_module/QUICKSTART.md b/Agent_module/QUICKSTART.md
new file mode 100644
index 0000000000..c3dd7baef5
--- /dev/null
+++ b/Agent_module/QUICKSTART.md
@@ -0,0 +1,414 @@
+# Agent Module — Quick Start
+
+This guide walks you through the most common usage patterns, from running a single agent to building a multi-agent pipeline with shared state.
+
+**Requirements**: Python 3.10+ — no third-party packages needed, stdlib only.
+
+---
+
+## Step 1 — Run a Built-in Agent
+
+The simplest starting point: instantiate one of the built-in agents and send it a message.
+
+```python
+# step1_single_agent.py
+import asyncio
+import time
+import sys, os
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from Framework import AgentMessage
+from Implementation import CalculatorAgent, WeatherAgent, TranslatorAgent
+
+async def main():
+ # --- Calculator ---
+ calc = CalculatorAgent()
+ response = await calc.process(AgentMessage(
+ content="(10 + 5) * 4",
+ sender="User",
+ timestamp=time.time()
+ ))
+ print(response.content) # (10 + 5) * 4 = 60
+
+ # --- Weather ---
+ weather = WeatherAgent()
+ response = await weather.process(AgentMessage(
+ content="Tokyo",
+ sender="User",
+ timestamp=time.time()
+ ))
+ print(response.content)
+
+ # --- Translator (target language via metadata) ---
+ translator = TranslatorAgent()
+ response = await translator.process(AgentMessage(
+ content="Good morning",
+ sender="User",
+ timestamp=time.time(),
+ metadata={"target_language": "french"}
+ ))
+ print(response.content)
+
+asyncio.run(main())
+```
+
+---
+
+## Step 2 — Write Your Own Agent
+
+Subclass `BaseAgent` and implement `_process_impl`. That is the only requirement.
+
+```python
+# step2_custom_agent.py
+import asyncio
+import time
+import sys, os
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from Framework import BaseAgent, AgentMessage
+
+class GreetingAgent(BaseAgent):
+ def __init__(self):
+ super().__init__(
+ name="GreetingAgent",
+ description="Returns a personalised greeting"
+ )
+
+ async def _process_impl(self, message: AgentMessage) -> AgentMessage:
+ name = message.metadata.get("user_name", "stranger")
+ return AgentMessage(
+ content=f"Hello, {name}! You said: '{message.content}'",
+ sender=self.name,
+ timestamp=time.time(),
+ message_type="greeting"
+ )
+
+async def main():
+ agent = GreetingAgent()
+ response = await agent.process(AgentMessage(
+ content="Nice to meet you",
+ sender="User",
+ timestamp=time.time(),
+ metadata={"user_name": "Alice"}
+ ))
+ print(response.content)
+ # Hello, Alice! You said: 'Nice to meet you'
+
+ print("Agent status:", agent.status.value) # idle
+ print("Messages processed:", len(agent.message_history)) # 1
+
+asyncio.run(main())
+```
+
+---
+
+## Step 3 — Manage Multiple Agents
+
+`AgentManager` handles registration, routing, and broadcast. Use it when you have more than one agent.
+
+```python
+# step3_manager.py
+import asyncio
+import sys, os
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from Manager import AgentManager
+from Implementation import CalculatorAgent, WeatherAgent, TranslatorAgent
+from Chat_Agent import ChatAgent
+
+async def main():
+ manager = AgentManager()
+ manager.register_agent(CalculatorAgent())
+ manager.register_agent(WeatherAgent())
+ manager.register_agent(TranslatorAgent())
+ manager.register_agent(ChatAgent())
+
+ # --- Unicast: send to one agent by name ---
+ response = await manager.send_message("CalculatorAgent", "99 / 3")
+ print(response.content) # 99 / 3 = 33.0
+
+ response = await manager.send_message("WeatherAgent", "London")
+ print(response.content)
+
+ # --- Broadcast: send to every agent concurrently ---
+ print("\n--- Broadcast ---")
+ responses = await manager.broadcast_message("Hello!")
+ for agent_name, resp in responses.items():
+ print(f" {agent_name}: {resp.content[:60]}")
+
+ # --- Capability discovery: find agents by keyword ---
+ print("\n--- Discovery ---")
+ print(manager.find_agent_by_capability("weather"))
+ print(manager.find_agent_by_capability("calculation"))
+
+ # --- System statistics ---
+ print("\n--- Stats ---")
+ stats = manager.get_system_stats()
+ for k, v in stats.items():
+ print(f" {k}: {v}")
+
+asyncio.run(main())
+```
+
+---
+
+## Step 4 — Share State Between Agents with ContextStore
+
+`ContextStore` is the recommended way for agents to exchange data. It is async, namespace-isolated, and keeps a full write history per key.
+
+```python
+# step4_context_store.py
+import asyncio
+import sys, os
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from ContextStore import ContextStore, ContextNamespace
+
+async def main():
+ store = ContextStore()
+ ns = ContextNamespace(store, "pipeline_v1")
+
+ # --- coordinator writes a task ---
+ await ns.put("task", {"id": "t1", "desc": "process sales data"}, writer="coordinator")
+ await ns.put("status", "pending", writer="coordinator")
+
+ # --- worker reads the task and updates state ---
+ task = await ns.get("task")
+ print("Task received:", task)
+
+ await ns.put("status", "in_progress", writer="worker")
+ await ns.put("result", {"rows": 5000, "errors": 0}, writer="worker", tags=["output"])
+ await ns.put("status", "completed", writer="worker")
+
+ # --- read the final status ---
+ print("Final status:", await ns.get("status"))
+
+ # --- inspect the full write history for "status" (newest first) ---
+ print("\nStatus history:")
+ for record in await ns.history("status"):
+ print(f" v{record.version} [{record.writer}] -> {record.value!r}")
+
+ # --- query all entries written by "worker" ---
+ print("\nWorker entries:")
+ for entry in await ns.query(writer="worker"):
+ print(f" {entry.current.key}: {entry.current.value}")
+
+asyncio.run(main())
+```
+
+---
+
+## Step 5 — React to Changes with Pub/Sub
+
+Subscribe a callback to be notified whenever a key is written.
+
+```python
+# step5_pubsub.py
+import asyncio
+import sys, os
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from ContextStore import ContextStore, ContextNamespace, ContextRecord
+
+store = ContextStore()
+ns = ContextNamespace(store, "pipeline_v1")
+
+# Callback can be sync or async
+async def on_status_change(key: str, record: ContextRecord) -> None:
+ print(f" [monitor] {key} -> {record.value!r} (writer: {record.writer})")
+
+async def main():
+ # Subscribe to specific keys
+ ns.subscribe("monitor", on_status_change, keys=["status"])
+
+ print("Writing status updates...")
+ await ns.put("status", "pending", writer="coordinator")
+ await ns.put("status", "in_progress", writer="worker")
+ await ns.put("status", "completed", writer="worker")
+
+ # The writer does not receive its own notifications.
+ # Subscribing with no keys list watches the entire namespace.
+ ns.subscribe("logger", lambda k, r: print(f" [logger] {k}={r.value}"))
+ await ns.put("result", {"score": 0.98}, writer="worker")
+
+asyncio.run(main())
+```
+
+Expected output:
+```
+Writing status updates...
+ [monitor] status -> 'pending' (writer: coordinator)
+ [monitor] status -> 'in_progress' (writer: worker)
+ [monitor] status -> 'completed' (writer: worker)
+ [logger] result={'score': 0.98}
+```
+
+---
+
+## Step 6 — Send Direct Messages Between Agents
+
+Use channels for point-to-point communication that does not go through the shared state.
+
+```python
+# step6_channels.py
+import asyncio
+import sys, os
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from ContextStore import ContextStore
+
+async def main():
+ store = ContextStore()
+
+ # coordinator sends two commands to worker
+ await store.send("coordinator", "worker", {"cmd": "run_report", "format": "pdf"})
+ await store.send("coordinator", "worker", {"cmd": "upload", "dest": "s3://bucket"})
+
+ # auditor sends a health check to worker
+ await store.send("auditor", "worker", {"cmd": "health_check"})
+
+ # worker drains its entire inbox (all senders, sorted by time)
+ inbox = await store.inbox("worker")
+ print(f"Worker has {len(inbox)} message(s):")
+ for msg in inbox:
+ print(f" from={msg.sender:<15} payload={msg.payload}")
+
+ # inbox is now empty
+ print("Inbox after drain:", await store.inbox("worker"))
+
+asyncio.run(main())
+```
+
+---
+
+## Step 7 — Write Time-Limited Data with TTL
+
+Entries with a TTL are automatically evicted after the specified number of seconds.
+
+```python
+# step7_ttl.py
+import asyncio
+import sys, os
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from ContextStore import ContextStore, ContextNamespace
+
+async def main():
+ store = ContextStore()
+ ns = ContextNamespace(store, "pipeline_v1")
+
+ # Write a heartbeat that expires in 3 seconds
+ await ns.put("heartbeat", {"alive": True, "load": 0.42}, writer="worker", ttl=3)
+
+ print("Immediate read:", await ns.get("heartbeat")) # {'alive': True, 'load': 0.42}
+
+ await asyncio.sleep(3.5)
+
+ print("After 3.5 s: ", await ns.get("heartbeat")) # None (expired)
+
+ # Optional: batch-remove any remaining stale entries
+ removed = await ns.cleanup_expired()
+ print(f"Cleaned up {removed} expired key(s)")
+
+asyncio.run(main())
+```
+
+---
+
+## Step 8 — Connect Agents to ContextStore
+
+Combine the agent framework with `ContextStore` to build a full pipeline.
+
+```python
+# step8_pipeline.py
+import asyncio
+import time
+import sys, os
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from Framework import BaseAgent, AgentMessage
+from ContextStore import ContextStore, ContextNamespace, ContextRecord
+
+store = ContextStore()
+ns = ContextNamespace(store, "pipeline")
+
+
+class CoordinatorAgent(BaseAgent):
+ async def _process_impl(self, message: AgentMessage) -> AgentMessage:
+ task = {"id": "t1", "desc": message.content}
+ await ns.put("task", task, writer=self.name)
+ await ns.put("status", "pending", writer=self.name)
+ return AgentMessage(content=f"Task assigned: {task['id']}",
+ sender=self.name, timestamp=time.time())
+
+
+class WorkerAgent(BaseAgent):
+ async def _process_impl(self, message: AgentMessage) -> AgentMessage:
+ task = await ns.get("task")
+ if task is None:
+ return AgentMessage(content="No task found",
+ sender=self.name, timestamp=time.time())
+
+ await ns.put("status", "in_progress", writer=self.name)
+ # Simulate work
+ await asyncio.sleep(0.1)
+ await ns.put("result", {"task_id": task["id"], "output": "done"},
+ writer=self.name, tags=["output"])
+ await ns.put("status", "completed", writer=self.name)
+ return AgentMessage(content="Work completed", sender=self.name, timestamp=time.time())
+
+
+async def main():
+ coordinator = CoordinatorAgent(name="coordinator")
+ worker = WorkerAgent(name="worker")
+
+ # Monitor all context changes
+ def log(key: str, record: ContextRecord) -> None:
+ print(f" [monitor] {key} = {record.value!r}")
+ ns.subscribe("monitor", log)
+
+ print("=== Coordinator assigns task ===")
+ r = await coordinator.process(AgentMessage(
+ content="Analyse Q1 sales", sender="orchestrator", timestamp=time.time()
+ ))
+ print(r.content)
+
+ print("\n=== Worker processes task ===")
+ r = await worker.process(AgentMessage(
+ content="start", sender="orchestrator", timestamp=time.time()
+ ))
+ print(r.content)
+
+ print("\n=== Final state ===")
+ print("status:", await ns.get("status"))
+ print("result:", await ns.get("result"))
+
+ print("\n=== Status history ===")
+ for rec in await ns.history("status"):
+ print(f" v{rec.version} [{rec.writer}] -> {rec.value!r}")
+
+asyncio.run(main())
+```
+
+---
+
+## Patterns at a Glance
+
+| Goal | What to use |
+|---|---|
+| Build a new agent | Subclass `BaseAgent`, implement `_process_impl` |
+| Run multiple agents | `AgentManager.register_agent()` + `send_message()` |
+| Share data between agents | `ContextStore` / `ContextNamespace` — `put()` / `get()` |
+| React to data changes | `ns.subscribe(agent_id, callback)` |
+| Send agent-to-agent commands | `store.send()` / `store.inbox()` |
+| Expire data automatically | `ns.put(..., ttl=<seconds>)` |
+| Audit what was written | `ns.history(key)` |
+| Filter entries by author/tag | `ns.query(writer=..., tags=[...])` |
+
+---
+
+## Legacy API (SharedContext)
+
+If you need the older synchronous API, use `SharedContext` + `ContextBridge` directly. The `SharedContext_Demo.py` file shows the full pattern. Note that `ContextBridge` loads its dependency using a custom `importlib` loader — refer to the demo's `_load()` helper for the correct usage.
+
+For all new development, use `ContextStore`.