forked from Rockachopa/Timmy-time-dashboard
This commit is contained in:
@@ -1 +0,0 @@
|
||||
"""Agent Core — Substrate-agnostic agent interface and base classes."""
|
||||
@@ -1,381 +0,0 @@
|
||||
"""TimAgent Interface — The substrate-agnostic agent contract.
|
||||
|
||||
This is the foundation for embodiment. Whether Timmy runs on:
|
||||
- A server with Ollama (today)
|
||||
- A Raspberry Pi with sensors
|
||||
- A Boston Dynamics Spot robot
|
||||
- A VR avatar
|
||||
|
||||
The interface remains constant. Implementation varies.
|
||||
|
||||
Architecture:
|
||||
perceive() → reason → act()
|
||||
↑ ↓
|
||||
←←← remember() ←←←←←←┘
|
||||
|
||||
All methods return effects that can be logged, audited, and replayed.
|
||||
"""
|
||||
|
||||
import uuid
|
||||
from abc import ABC, abstractmethod
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import UTC, datetime
|
||||
from enum import Enum, auto
|
||||
from typing import Any
|
||||
|
||||
|
||||
class PerceptionType(Enum):
|
||||
"""Types of sensory input an agent can receive."""
|
||||
|
||||
TEXT = auto() # Natural language
|
||||
IMAGE = auto() # Visual input
|
||||
AUDIO = auto() # Sound/speech
|
||||
SENSOR = auto() # Temperature, distance, etc.
|
||||
MOTION = auto() # Accelerometer, gyroscope
|
||||
NETWORK = auto() # API calls, messages
|
||||
INTERNAL = auto() # Self-monitoring (battery, temp)
|
||||
|
||||
|
||||
class ActionType(Enum):
|
||||
"""Types of actions an agent can perform."""
|
||||
|
||||
TEXT = auto() # Generate text response
|
||||
SPEAK = auto() # Text-to-speech
|
||||
MOVE = auto() # Physical movement
|
||||
GRIP = auto() # Manipulate objects
|
||||
CALL = auto() # API/network call
|
||||
EMIT = auto() # Signal/light/sound
|
||||
SLEEP = auto() # Power management
|
||||
|
||||
|
||||
class AgentCapability(Enum):
|
||||
"""High-level capabilities a TimAgent may possess."""
|
||||
|
||||
REASONING = "reasoning"
|
||||
CODING = "coding"
|
||||
WRITING = "writing"
|
||||
ANALYSIS = "analysis"
|
||||
VISION = "vision"
|
||||
SPEECH = "speech"
|
||||
NAVIGATION = "navigation"
|
||||
MANIPULATION = "manipulation"
|
||||
LEARNING = "learning"
|
||||
COMMUNICATION = "communication"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class AgentIdentity:
|
||||
"""Immutable identity for an agent instance.
|
||||
|
||||
This persists across sessions and substrates. If Timmy moves
|
||||
from cloud to robot, the identity follows.
|
||||
"""
|
||||
|
||||
id: str
|
||||
name: str
|
||||
version: str
|
||||
created_at: str = field(default_factory=lambda: datetime.now(UTC).isoformat())
|
||||
|
||||
@classmethod
|
||||
def generate(cls, name: str, version: str = "1.0.0") -> "AgentIdentity":
|
||||
"""Generate a new unique identity."""
|
||||
return cls(
|
||||
id=str(uuid.uuid4()),
|
||||
name=name,
|
||||
version=version,
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Perception:
|
||||
"""A sensory input to the agent.
|
||||
|
||||
Substrate-agnostic representation. A camera image and a
|
||||
LiDAR point cloud are both Perception instances.
|
||||
"""
|
||||
|
||||
type: PerceptionType
|
||||
data: Any # Content depends on type
|
||||
timestamp: str = field(default_factory=lambda: datetime.now(UTC).isoformat())
|
||||
source: str = "unknown" # e.g., "camera_1", "microphone", "user_input"
|
||||
metadata: dict = field(default_factory=dict)
|
||||
|
||||
@classmethod
|
||||
def text(cls, content: str, source: str = "user") -> "Perception":
|
||||
"""Factory for text perception."""
|
||||
return cls(
|
||||
type=PerceptionType.TEXT,
|
||||
data=content,
|
||||
source=source,
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def sensor(cls, kind: str, value: float, unit: str = "") -> "Perception":
|
||||
"""Factory for sensor readings."""
|
||||
return cls(
|
||||
type=PerceptionType.SENSOR,
|
||||
data={"kind": kind, "value": value, "unit": unit},
|
||||
source=f"sensor_{kind}",
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Action:
|
||||
"""An action the agent intends to perform.
|
||||
|
||||
Actions are effects — they describe what should happen,
|
||||
not how. The substrate implements the "how."
|
||||
"""
|
||||
|
||||
type: ActionType
|
||||
payload: Any # Action-specific data
|
||||
timestamp: str = field(default_factory=lambda: datetime.now(UTC).isoformat())
|
||||
confidence: float = 1.0 # 0-1, agent's certainty
|
||||
deadline: str | None = None # When action must complete
|
||||
|
||||
@classmethod
|
||||
def respond(cls, text: str, confidence: float = 1.0) -> "Action":
|
||||
"""Factory for text response action."""
|
||||
return cls(
|
||||
type=ActionType.TEXT,
|
||||
payload=text,
|
||||
confidence=confidence,
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def move(cls, vector: tuple[float, float, float], speed: float = 1.0) -> "Action":
|
||||
"""Factory for movement action (x, y, z meters)."""
|
||||
return cls(
|
||||
type=ActionType.MOVE,
|
||||
payload={"vector": vector, "speed": speed},
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Memory:
|
||||
"""A stored experience or fact.
|
||||
|
||||
Memories are substrate-agnostic. A conversation history
|
||||
and a video recording are both Memory instances.
|
||||
"""
|
||||
|
||||
id: str
|
||||
content: Any
|
||||
created_at: str
|
||||
access_count: int = 0
|
||||
last_accessed: str | None = None
|
||||
importance: float = 0.5 # 0-1, for pruning decisions
|
||||
tags: list[str] = field(default_factory=list)
|
||||
|
||||
def touch(self) -> None:
|
||||
"""Mark memory as accessed."""
|
||||
self.access_count += 1
|
||||
self.last_accessed = datetime.now(UTC).isoformat()
|
||||
|
||||
|
||||
@dataclass
|
||||
class Communication:
|
||||
"""A message to/from another agent or human."""
|
||||
|
||||
sender: str
|
||||
recipient: str
|
||||
content: Any
|
||||
timestamp: str = field(default_factory=lambda: datetime.now(UTC).isoformat())
|
||||
protocol: str = "direct" # e.g., "http", "websocket", "speech"
|
||||
encrypted: bool = False
|
||||
|
||||
|
||||
class TimAgent(ABC):
|
||||
"""Abstract base class for all Timmy agent implementations.
|
||||
|
||||
This is the substrate-agnostic interface. Implementations:
|
||||
- OllamaAgent: LLM-based reasoning (today)
|
||||
- RobotAgent: Physical embodiment (future)
|
||||
- SimulationAgent: Virtual environment (future)
|
||||
|
||||
Usage:
|
||||
agent = OllamaAgent(identity) # Today's implementation
|
||||
|
||||
perception = Perception.text("Hello Timmy")
|
||||
memory = agent.perceive(perception)
|
||||
|
||||
action = agent.reason("How should I respond?")
|
||||
result = agent.act(action)
|
||||
|
||||
agent.remember(memory) # Store for future
|
||||
"""
|
||||
|
||||
def __init__(self, identity: AgentIdentity) -> None:
|
||||
self._identity = identity
|
||||
self._capabilities: set[AgentCapability] = set()
|
||||
self._state: dict[str, Any] = {}
|
||||
|
||||
@property
|
||||
def identity(self) -> AgentIdentity:
|
||||
"""Return this agent's immutable identity."""
|
||||
return self._identity
|
||||
|
||||
@property
|
||||
def capabilities(self) -> set[AgentCapability]:
|
||||
"""Return set of supported capabilities."""
|
||||
return self._capabilities.copy()
|
||||
|
||||
def has_capability(self, capability: AgentCapability) -> bool:
|
||||
"""Check if agent supports a capability."""
|
||||
return capability in self._capabilities
|
||||
|
||||
@abstractmethod
|
||||
def perceive(self, perception: Perception) -> Memory:
|
||||
"""Process sensory input and create a memory.
|
||||
|
||||
This is the entry point for all agent interaction.
|
||||
A text message, camera frame, or temperature reading
|
||||
all enter through perceive().
|
||||
|
||||
Args:
|
||||
perception: Sensory input
|
||||
|
||||
Returns:
|
||||
Memory: Stored representation of the perception
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def reason(self, query: str, context: list[Memory]) -> Action:
|
||||
"""Reason about a situation and decide on action.
|
||||
|
||||
This is where "thinking" happens. The agent uses its
|
||||
substrate-appropriate reasoning (LLM, neural net, rules)
|
||||
to decide what to do.
|
||||
|
||||
Args:
|
||||
query: What to reason about
|
||||
context: Relevant memories for context
|
||||
|
||||
Returns:
|
||||
Action: What the agent decides to do
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def act(self, action: Action) -> Any:
|
||||
"""Execute an action in the substrate.
|
||||
|
||||
This is where the abstract action becomes concrete:
|
||||
- TEXT → Generate LLM response
|
||||
- MOVE → Send motor commands
|
||||
- SPEAK → Call TTS engine
|
||||
|
||||
Args:
|
||||
action: The action to execute
|
||||
|
||||
Returns:
|
||||
Result of the action (substrate-specific)
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def remember(self, memory: Memory) -> None:
|
||||
"""Store a memory for future retrieval.
|
||||
|
||||
The storage mechanism depends on substrate:
|
||||
- Cloud: SQLite, vector DB
|
||||
- Robot: Local flash storage
|
||||
- Hybrid: Synced with conflict resolution
|
||||
|
||||
Args:
|
||||
memory: Experience to store
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def recall(self, query: str, limit: int = 5) -> list[Memory]:
|
||||
"""Retrieve relevant memories.
|
||||
|
||||
Args:
|
||||
query: What to search for
|
||||
limit: Maximum memories to return
|
||||
|
||||
Returns:
|
||||
List of relevant memories, sorted by relevance
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def communicate(self, message: Communication) -> bool:
|
||||
"""Send/receive communication with another agent.
|
||||
|
||||
Args:
|
||||
message: Message to send
|
||||
|
||||
Returns:
|
||||
True if communication succeeded
|
||||
"""
|
||||
pass
|
||||
|
||||
def get_state(self) -> dict[str, Any]:
|
||||
"""Get current agent state for monitoring/debugging."""
|
||||
return {
|
||||
"identity": self._identity,
|
||||
"capabilities": list(self._capabilities),
|
||||
"state": self._state.copy(),
|
||||
}
|
||||
|
||||
def shutdown(self) -> None: # noqa: B027
|
||||
"""Graceful shutdown. Persist state, close connections."""
|
||||
# Override in subclass for cleanup
|
||||
|
||||
|
||||
class AgentEffect:
|
||||
"""Log entry for agent actions — for audit and replay.
|
||||
|
||||
The complete history of an agent's life can be captured
|
||||
as a sequence of AgentEffects. This enables:
|
||||
- Debugging: What did the agent see and do?
|
||||
- Audit: Why did it make that decision?
|
||||
- Replay: Reconstruct agent state from log
|
||||
- Training: Learn from agent experiences
|
||||
"""
|
||||
|
||||
def __init__(self, log_path: str | None = None) -> None:
|
||||
self._effects: list[dict] = []
|
||||
self._log_path = log_path
|
||||
|
||||
def log_perceive(self, perception: Perception, memory_id: str) -> None:
|
||||
"""Log a perception event."""
|
||||
self._effects.append(
|
||||
{
|
||||
"type": "perceive",
|
||||
"perception_type": perception.type.name,
|
||||
"source": perception.source,
|
||||
"memory_id": memory_id,
|
||||
"timestamp": datetime.now(UTC).isoformat(),
|
||||
}
|
||||
)
|
||||
|
||||
def log_reason(self, query: str, action_type: ActionType) -> None:
|
||||
"""Log a reasoning event."""
|
||||
self._effects.append(
|
||||
{
|
||||
"type": "reason",
|
||||
"query": query,
|
||||
"action_type": action_type.name,
|
||||
"timestamp": datetime.now(UTC).isoformat(),
|
||||
}
|
||||
)
|
||||
|
||||
def log_act(self, action: Action, result: Any) -> None:
|
||||
"""Log an action event."""
|
||||
self._effects.append(
|
||||
{
|
||||
"type": "act",
|
||||
"action_type": action.type.name,
|
||||
"confidence": action.confidence,
|
||||
"result_type": type(result).__name__,
|
||||
"timestamp": datetime.now(UTC).isoformat(),
|
||||
}
|
||||
)
|
||||
|
||||
def export(self) -> list[dict]:
|
||||
"""Export effect log for analysis."""
|
||||
return self._effects.copy()
|
||||
@@ -1,343 +0,0 @@
|
||||
"""Functional tests for agent_core — interface.
|
||||
|
||||
Covers the substrate-agnostic agent contract (data classes, enums,
|
||||
factory methods, abstract enforcement).
|
||||
"""
|
||||
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
|
||||
from timmy.agent_core.interface import (
|
||||
Action,
|
||||
ActionType,
|
||||
AgentCapability,
|
||||
AgentEffect,
|
||||
AgentIdentity,
|
||||
Communication,
|
||||
Memory,
|
||||
Perception,
|
||||
PerceptionType,
|
||||
TimAgent,
|
||||
)
|
||||
|
||||
# ── AgentIdentity ─────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestAgentIdentity:
|
||||
def test_generate_creates_uuid(self):
|
||||
identity = AgentIdentity.generate("Timmy")
|
||||
assert identity.name == "Timmy"
|
||||
uuid.UUID(identity.id) # raises on invalid
|
||||
|
||||
def test_generate_default_version(self):
|
||||
identity = AgentIdentity.generate("Timmy")
|
||||
assert identity.version == "1.0.0"
|
||||
|
||||
def test_generate_custom_version(self):
|
||||
identity = AgentIdentity.generate("Timmy", version="2.0.0")
|
||||
assert identity.version == "2.0.0"
|
||||
|
||||
def test_frozen_identity(self):
|
||||
identity = AgentIdentity.generate("Timmy")
|
||||
with pytest.raises(AttributeError):
|
||||
identity.name = "Other"
|
||||
|
||||
def test_created_at_populated(self):
|
||||
identity = AgentIdentity.generate("Timmy")
|
||||
assert identity.created_at # not empty
|
||||
assert "T" in identity.created_at # ISO format
|
||||
|
||||
def test_two_identities_differ(self):
|
||||
a = AgentIdentity.generate("A")
|
||||
b = AgentIdentity.generate("B")
|
||||
assert a.id != b.id
|
||||
|
||||
|
||||
# ── Perception ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestPerception:
|
||||
def test_text_factory(self):
|
||||
p = Perception.text("hello")
|
||||
assert p.type == PerceptionType.TEXT
|
||||
assert p.data == "hello"
|
||||
assert p.source == "user"
|
||||
|
||||
def test_text_factory_custom_source(self):
|
||||
p = Perception.text("hello", source="api")
|
||||
assert p.source == "api"
|
||||
|
||||
def test_sensor_factory(self):
|
||||
p = Perception.sensor("temperature", 22.5, "°C")
|
||||
assert p.type == PerceptionType.SENSOR
|
||||
assert p.data["kind"] == "temperature"
|
||||
assert p.data["value"] == 22.5
|
||||
assert p.data["unit"] == "°C"
|
||||
assert p.source == "sensor_temperature"
|
||||
|
||||
def test_timestamp_auto_populated(self):
|
||||
p = Perception.text("hi")
|
||||
assert p.timestamp
|
||||
assert "T" in p.timestamp
|
||||
|
||||
def test_metadata_defaults_empty(self):
|
||||
p = Perception.text("hi")
|
||||
assert p.metadata == {}
|
||||
|
||||
|
||||
# ── Action ────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestAction:
|
||||
def test_respond_factory(self):
|
||||
a = Action.respond("Hello!")
|
||||
assert a.type == ActionType.TEXT
|
||||
assert a.payload == "Hello!"
|
||||
assert a.confidence == 1.0
|
||||
|
||||
def test_respond_with_confidence(self):
|
||||
a = Action.respond("Maybe", confidence=0.5)
|
||||
assert a.confidence == 0.5
|
||||
|
||||
def test_move_factory(self):
|
||||
a = Action.move((1.0, 2.0, 3.0), speed=0.5)
|
||||
assert a.type == ActionType.MOVE
|
||||
assert a.payload["vector"] == (1.0, 2.0, 3.0)
|
||||
assert a.payload["speed"] == 0.5
|
||||
|
||||
def test_move_default_speed(self):
|
||||
a = Action.move((0, 0, 0))
|
||||
assert a.payload["speed"] == 1.0
|
||||
|
||||
def test_deadline_defaults_none(self):
|
||||
a = Action.respond("test")
|
||||
assert a.deadline is None
|
||||
|
||||
|
||||
# ── Memory ────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestMemory:
|
||||
def test_touch_increments(self):
|
||||
m = Memory(id="m1", content="hello", created_at="2025-01-01T00:00:00Z")
|
||||
assert m.access_count == 0
|
||||
m.touch()
|
||||
assert m.access_count == 1
|
||||
m.touch()
|
||||
assert m.access_count == 2
|
||||
|
||||
def test_touch_sets_last_accessed(self):
|
||||
m = Memory(id="m1", content="hello", created_at="2025-01-01T00:00:00Z")
|
||||
assert m.last_accessed is None
|
||||
m.touch()
|
||||
assert m.last_accessed is not None
|
||||
|
||||
def test_default_importance(self):
|
||||
m = Memory(id="m1", content="x", created_at="now")
|
||||
assert m.importance == 0.5
|
||||
|
||||
def test_tags_default_empty(self):
|
||||
m = Memory(id="m1", content="x", created_at="now")
|
||||
assert m.tags == []
|
||||
|
||||
|
||||
# ── Communication ─────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestCommunication:
|
||||
def test_defaults(self):
|
||||
c = Communication(sender="A", recipient="B", content="hi")
|
||||
assert c.protocol == "direct"
|
||||
assert c.encrypted is False
|
||||
assert c.timestamp # auto-populated
|
||||
|
||||
|
||||
# ── TimAgent abstract enforcement ─────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestTimAgentABC:
|
||||
def test_cannot_instantiate_abstract(self):
|
||||
with pytest.raises(TypeError):
|
||||
TimAgent(AgentIdentity.generate("X"))
|
||||
|
||||
def test_concrete_subclass_works(self):
|
||||
class Dummy(TimAgent):
|
||||
def perceive(self, p):
|
||||
return Memory(id="1", content=p.data, created_at="")
|
||||
|
||||
def reason(self, q, c):
|
||||
return Action.respond(q)
|
||||
|
||||
def act(self, a):
|
||||
return a.payload
|
||||
|
||||
def remember(self, m):
|
||||
pass
|
||||
|
||||
def recall(self, q, limit=5):
|
||||
return []
|
||||
|
||||
def communicate(self, m):
|
||||
return True
|
||||
|
||||
d = Dummy(AgentIdentity.generate("Dummy"))
|
||||
assert d.identity.name == "Dummy"
|
||||
assert d.capabilities == set()
|
||||
|
||||
def test_has_capability(self):
|
||||
class Dummy(TimAgent):
|
||||
def perceive(self, p):
|
||||
pass
|
||||
|
||||
def reason(self, q, c):
|
||||
pass
|
||||
|
||||
def act(self, a):
|
||||
pass
|
||||
|
||||
def remember(self, m):
|
||||
pass
|
||||
|
||||
def recall(self, q, limit=5):
|
||||
return []
|
||||
|
||||
def communicate(self, m):
|
||||
return True
|
||||
|
||||
d = Dummy(AgentIdentity.generate("D"))
|
||||
d._capabilities.add(AgentCapability.REASONING)
|
||||
assert d.has_capability(AgentCapability.REASONING)
|
||||
assert not d.has_capability(AgentCapability.VISION)
|
||||
|
||||
def test_capabilities_returns_copy(self):
|
||||
class Dummy(TimAgent):
|
||||
def perceive(self, p):
|
||||
pass
|
||||
|
||||
def reason(self, q, c):
|
||||
pass
|
||||
|
||||
def act(self, a):
|
||||
pass
|
||||
|
||||
def remember(self, m):
|
||||
pass
|
||||
|
||||
def recall(self, q, limit=5):
|
||||
return []
|
||||
|
||||
def communicate(self, m):
|
||||
return True
|
||||
|
||||
d = Dummy(AgentIdentity.generate("D"))
|
||||
caps = d.capabilities
|
||||
caps.add(AgentCapability.VISION)
|
||||
assert AgentCapability.VISION not in d.capabilities
|
||||
|
||||
def test_get_state(self):
|
||||
class Dummy(TimAgent):
|
||||
def perceive(self, p):
|
||||
pass
|
||||
|
||||
def reason(self, q, c):
|
||||
pass
|
||||
|
||||
def act(self, a):
|
||||
pass
|
||||
|
||||
def remember(self, m):
|
||||
pass
|
||||
|
||||
def recall(self, q, limit=5):
|
||||
return []
|
||||
|
||||
def communicate(self, m):
|
||||
return True
|
||||
|
||||
d = Dummy(AgentIdentity.generate("D"))
|
||||
state = d.get_state()
|
||||
assert "identity" in state
|
||||
assert "capabilities" in state
|
||||
assert "state" in state
|
||||
|
||||
def test_shutdown_does_not_raise(self):
|
||||
class Dummy(TimAgent):
|
||||
def perceive(self, p):
|
||||
pass
|
||||
|
||||
def reason(self, q, c):
|
||||
pass
|
||||
|
||||
def act(self, a):
|
||||
pass
|
||||
|
||||
def remember(self, m):
|
||||
pass
|
||||
|
||||
def recall(self, q, limit=5):
|
||||
return []
|
||||
|
||||
def communicate(self, m):
|
||||
return True
|
||||
|
||||
d = Dummy(AgentIdentity.generate("D"))
|
||||
d.shutdown() # should not raise
|
||||
|
||||
|
||||
# ── AgentEffect ───────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestAgentEffect:
|
||||
def test_empty_export(self):
|
||||
effect = AgentEffect()
|
||||
assert effect.export() == []
|
||||
|
||||
def test_log_perceive(self):
|
||||
effect = AgentEffect()
|
||||
p = Perception.text("test input")
|
||||
effect.log_perceive(p, "mem_0")
|
||||
log = effect.export()
|
||||
assert len(log) == 1
|
||||
assert log[0]["type"] == "perceive"
|
||||
assert log[0]["perception_type"] == "TEXT"
|
||||
assert log[0]["memory_id"] == "mem_0"
|
||||
assert "timestamp" in log[0]
|
||||
|
||||
def test_log_reason(self):
|
||||
effect = AgentEffect()
|
||||
effect.log_reason("How to help?", ActionType.TEXT)
|
||||
log = effect.export()
|
||||
assert len(log) == 1
|
||||
assert log[0]["type"] == "reason"
|
||||
assert log[0]["query"] == "How to help?"
|
||||
assert log[0]["action_type"] == "TEXT"
|
||||
|
||||
def test_log_act(self):
|
||||
effect = AgentEffect()
|
||||
action = Action.respond("Hello!")
|
||||
effect.log_act(action, "Hello!")
|
||||
log = effect.export()
|
||||
assert len(log) == 1
|
||||
assert log[0]["type"] == "act"
|
||||
assert log[0]["confidence"] == 1.0
|
||||
assert log[0]["result_type"] == "str"
|
||||
|
||||
def test_export_returns_copy(self):
|
||||
effect = AgentEffect()
|
||||
effect.log_reason("q", ActionType.TEXT)
|
||||
exported = effect.export()
|
||||
exported.clear()
|
||||
assert len(effect.export()) == 1
|
||||
|
||||
def test_full_audit_trail(self):
|
||||
effect = AgentEffect()
|
||||
p = Perception.text("input")
|
||||
effect.log_perceive(p, "m0")
|
||||
effect.log_reason("what now?", ActionType.TEXT)
|
||||
action = Action.respond("response")
|
||||
effect.log_act(action, "response")
|
||||
log = effect.export()
|
||||
assert len(log) == 3
|
||||
types = [e["type"] for e in log]
|
||||
assert types == ["perceive", "reason", "act"]
|
||||
Reference in New Issue
Block a user