Spaces:
Running
Running
File size: 2,369 Bytes
f4542ce | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 | """
React -- decides, given an agent's current action (if any) and what it just
perceived, whether to keep executing that action or interrupt and replan.
Design notes
------------
All calls into this module are cheap heuristic checks with no LLM round-trip:
- no current action yet -> always replan
- current action's end_tick has passed -> always replan
- mid-action, nothing new perceived this tick -> always continue
The LLM-based decision layer has moved to brain.decide_tick(), which runs
only when the perceive phase detects novel observations.
"""
from __future__ import annotations
from typing import Any, Dict, List, Optional
from pydantic import BaseModel
from src.core.log import get_logger
from src.core.world_state import CurrentAction
from src.core.perceive import Observation
logger = get_logger(__name__)
class ReactionDecision(BaseModel):
should_replan: bool
reason: str
def decide_reaction(
agent_id: str,
persona: Dict[str, Any],
current_action: Optional[CurrentAction],
observations: List[Observation],
memories: List[str],
tick: int,
) -> ReactionDecision:
"""Cheap heuristic decision: no LLM involved."""
if current_action is None:
return ReactionDecision(should_replan=True, reason="no current action to continue")
if current_action.is_finished(tick):
return ReactionDecision(should_replan=True, reason="current action has finished")
if not observations:
return ReactionDecision(should_replan=False, reason="mid-action, nothing new perceived")
# Mid-action with observations: cheap heuristic — always continue.
# The brain's decide_tick (LLM) will handle novel observation decisions.
return ReactionDecision(should_replan=False, reason="mid-action, continuing current action")
# Standalone sanity check (doesn't hit the network)
if __name__ == "__main__":
action = CurrentAction(description="sleeping", start_tick=0, end_tick=60)
r1 = decide_reaction("a", {}, None, [], [], tick=0)
assert r1.should_replan is True
r2 = decide_reaction("a", {}, action, [], [], tick=61)
assert r2.should_replan is True
r3 = decide_reaction("a", {}, action, [], [], tick=30)
assert r3.should_replan is False
print("react.py cheap-path sanity checks passed (LLM branch not exercised here).") |