AI代理感知框架

Sources (sensors/APIs) → Cortex (perception) → Your memory/reasoning layer

pip install cortex-agent
# or from source:
git clone https://github.com/tsubasa-rsrch/cortex.git
cd cortex && pip install -e .

from cortex import CortexConfig, set_config
from cortex import HabituationFilter, CircadianRhythm, DecisionEngine, Event

# Configure (optional — defaults to ~/.cortex/)
set_config(CortexConfig(data_dir="/tmp/my_agent", name="demo"))

# 1. Filter noise with habituation
hab = HabituationFilter(base_threshold=15.0)

should_alert, reason = hab.should_notify("camera_1", 25.0)
print(f"Alert: {should_alert} — {reason}")
# Alert: True — Motion (alert, value=25.0 >= threshold=15.0)

should_alert, reason = hab.should_notify("camera_1", 16.0)
print(f"Alert: {should_alert} — {reason}")
# Alert: False — Cooldown (2s < 60s)

# 2. Adapt behavior to time of day
circadian = CircadianRhythm()
status = circadian.check_and_update()
print(f"Mode: {status['mode'].value}, Changed: {status['changed']}")

# 3. Decide what to do
engine = DecisionEngine()
events = [
    Event(source="camera", type="motion", content="Movement detected", priority=8),
    Event(source="api", type="message", content="Hello", priority=3),
]
action = engine.decide(events)
print(f"Action: {action.name} — {action.description}")
# Processes highest-priority event first

hab = HabituationFilter(
    cooldown=60.0,          # Min seconds between alerts per source
    window=300.0,           # Habituation counting window
    habituate_count=3,      # Detections before habituation kicks in
    habituated_mult=2.0,    # Threshold multiplier when habituated
    orienting_mult=2.0,     # Multiplier for orienting response
    base_threshold=15.0,    # Base detection threshold
)
should_alert, reason = hab.should_notify("sensor_1", value=25.0)

circadian = CircadianRhythm()
result = circadian.check_and_update()

# Override defaults with your own suggestions
custom = CircadianRhythm(suggestions={
    "morning": [{"text": "Check inbox", "icon": "mail"}],
    "afternoon": [{"text": "Write code", "icon": "laptop"}],
    "evening": [{"text": "Review PRs", "icon": "search"}],
    "night": [{"text": "Run backups", "icon": "database"}],
})

from cortex import Scheduler

scheduler = Scheduler()
scheduler.register("health_check", interval_seconds=300, callback=check_health)
scheduler.register("sync_data", interval_seconds=3600, callback=sync)

# In your main loop:
results = scheduler.check_and_run()  # Runs any tasks that are due
status = scheduler.get_status()       # Human-readable timing info

from cortex import NotificationQueue

nq = NotificationQueue()
nq.push("alert", "Motion detected in lobby", priority="urgent")
nq.push("message", "New email from alice@example.com")

unread = nq.get_unread()
print(nq.format())
# Notifications (2):
#   !!🔔 [09:15] Motion detected in lobby
#   💬 [09:16] New email from alice@example.com

nq.mark_all_read()

from cortex import TimestampLog

tl = TimestampLog()
tl.start_task("deploy v2.1")
# ... work ...
tl.checkpoint("tests passed")
# ... more work ...
result = tl.end_task("deployed successfully")
print(f"Took {result['elapsed_min']:.1f} minutes")

status = tl.get_status()  # Current task, time elapsed, checkpoints

from cortex import DecisionEngine, Action, Event

# Custom event handler
def handle_camera(event):
    return Action("investigate", f"Check {event.content}")

engine = DecisionEngine(
    activities=[
        {"name": "explore", "description": "Browse the web", "weight": 3.0},
        {"name": "review", "description": "Review recent logs", "weight": 1.0},
    ],
    event_handlers={"camera": handle_camera},
)

# With events: routes to handler
action = engine.decide([Event(source="camera", type="motion", content="lobby", priority=7)])

# Without events: picks a random activity
action = engine.decide([])
result = action.execute()  # Runs the handler if one is attached

from cortex import BaseSource, Event

class SlackSource(BaseSource):
    @property
    def name(self) -> str:
        return "slack"

    def check(self) -> list:
        # Your polling logic here
        messages = self._fetch_new_messages()
        return [
            Event(
                source=self.name,
                type="message",
                content=msg["text"],
                author=msg["user"],
                priority=7 if "@agent" in msg["text"] else 3,
            )
            for msg in messages
        ]

pip install cortex-agent[reachy]

from reachy_mini import ReachyMini
from cortex.sources.reachy import ReachyCameraSource, ReachyAudioSource, ReachyIMUSource
from cortex.sources.vision import VisionSource

mini = ReachyMini(connection_mode="localhost_only")
camera = ReachyCameraSource(mini, diff_threshold=15.0, min_changed_ratio=0.0668)
audio = ReachyAudioSource(mini, energy_threshold=0.01)
imu = ReachyIMUSource(mini, accel_threshold=2.0)
vision = VisionSource(mini.media.get_frame)  # optional YOLO classification

# Poll in your main loop
events = camera.check() + audio.check() + imu.check() + vision.check()
action = engine.decide(events)

# Start mockup simulator
reachy-mini-daemon --mockup-sim --deactivate-audio --localhost-only

# Demo mode (scripted)
python examples/reachy_bridge.py

# Live mode (real-time perception loop)
python examples/reachy_bridge.py --live --interval 0.5

from cortex.bridges.elasticsearch import CortexElasticBridge, ESConfig

# Mock mode (no ES cluster needed)
bridge = CortexElasticBridge()

# Production mode
bridge = CortexElasticBridge(es_config=ESConfig(
    es_url="https://your-cluster.es.io:443",
    api_key="your-api-key",
    index_prefix="cortex-events",
    mock_mode=False,
))

# Filter and index an event (only novel stimuli pass through)
from cortex import Event
event = Event(source="camera", type="motion", content="Movement in lobby", priority=8,
              raw_data={"diff": 25.0})
result = bridge.index_event(event)  # Returns IndexedEvent or None

# Build Agent Builder context with circadian awareness
context = bridge.get_agent_context()
prompt = bridge.build_agent_system_prompt("You are a security agent.")

from cortex.bridges.gemini import CortexGeminiBridge, GeminiConfig

# Mock mode (no API key needed)
bridge = CortexGeminiBridge()

# Production mode
bridge = CortexGeminiBridge(gemini_config=GeminiConfig(
    api_key="your-gemini-api-key",
    model="gemini-3-flash-preview",
    mock_mode=False,
))

# Full pipeline: filter → reason → act
from cortex import Event
events = [
    Event(source="camera", type="motion", content="Person at door", priority=8,
          raw_data={"diff": 30.0}),
    Event(source="mic", type="audio", content="Doorbell", priority=6,
          raw_data={"diff": 20.0}),
]
result = bridge.perceive_and_reason(events)
if result:
    print(f"Action: {result.action} (confidence: {result.confidence:.0%})")
    print(f"Reasoning: {result.reasoning}")

# Interactive queries with full perception context
response = bridge.reason_about_context("What's happening right now?")

# Check efficiency stats
stats = bridge.get_stats()
print(f"Filter rate: {stats['filter_rate']}")  # e.g. "60.0%"

Sensors → Cortex (perception) → Gemini 3 (reasoning) → Actions
          habituation filter     contextual reasoning    alert
          circadian rhythm       planning & decisions    investigate
          priority assessment    natural language         log

# Mock mode (no API key)
python examples/gemini_cognitive_agent.py

# Real API
GEMINI_API_KEY=your-key python examples/gemini_cognitive_agent.py

# Install
pip install cortex-agent

# Add to Claude Code
claude mcp add cortex-perception -- python -m cortex.mcp_server

{
    "mcpServers": {
        "cortex-perception": {
            "command": "python3",
            "args": ["-m", "cortex.mcp_server"]
        }
    }
}

┌──────────────────────────────────────────────────────────────┐
│                       Your Agent                              │
│                                                               │
│  ┌──────────┐  ┌────────────┐  ┌──────────────────────────┐ │
│  │ Sources   │→│   Cortex    │→│  Memory / Reasoning       │ │
│  │ (sensors) │  │ (filtering) │  │  (Cognee, LangChain, …)  │ │
│  └──────────┘  └────────────┘  └──────────────────────────┘ │
│       │              │                    │                    │
│  BaseSource     HabituationFilter    ┌───────────────┐      │
│  Event          CircadianRhythm      │  Body / API    │      │
│  ReachyCamera   Scheduler            │  (ReachyMini,  │      │
│  ReachyAudio    NotificationQueue    │   Slack, …)    │      │
│  ReachyIMU      TimestampLog         └───────────────┘      │
│                 DecisionEngine                               │
│                      │                                        │
│              ┌───────────────┐                                │
│              │   Bridges      │                                │
│              │  Gemini 3      │                                │
│              │  Elasticsearch │                                │
│              │  MCP Server    │                                │
│              └───────────────┘                                │
└──────────────────────────────────────────────────────────────┘

from cortex import CortexConfig, set_config

config = CortexConfig(
    data_dir="/path/to/state",  # Where state files are stored (default: ~/.cortex/)
    name="my-agent",            # Agent identifier
)
set_config(config)  # Sets the global config used by all modules

hab = HabituationFilter()                        # Uses global config
scheduler = Scheduler(config=my_custom_config)    # Uses its own config

pip install pytest
python -m pytest tests/ -v

# Replay historical data with colored output
python examples/replay_demo.py          # Uses real event log if available
python examples/replay_demo.py --verbose # Show individual filter decisions

# Live real-time monitoring (watches daemon event log)
python examples/live_dashboard.py           # Full-screen mode
python examples/live_dashboard.py --compact # Streaming mode