feat: Implement enhanced action economy for LLM agent orchestration (#156)
- Add action economy system with free (LOOK, SPEAK) vs turn-ending (GO, WAIT, TAKE) actions - Implement LOOK action with detailed descriptions for doors, objects, entities, directions - Add SPEAK/ANNOUNCE speech system with room-wide and proximity-based message delivery - Create multi-tile pathing with FOV interrupt detection (path cancels when new entity visible) - Implement TAKE action with adjacency requirement and clear error messages - Add conversation history and error feedback loop so agents learn from failed actions - Create structured simulation logging for offline viewer replay - Document offline viewer requirements in OFFLINE_VIEWER_SPEC.md - Fix import path in 1_multi_agent_demo.py for standalone execution 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
parent
85e90088d5
commit
335efc5514
6 changed files with 2232 additions and 2 deletions
302
tests/vllm_demo/action_economy.py
Normal file
302
tests/vllm_demo/action_economy.py
Normal file
|
|
@ -0,0 +1,302 @@
|
|||
"""
|
||||
Action Economy System
|
||||
=====================
|
||||
|
||||
Defines which actions consume turns and which are free.
|
||||
Manages multi-tile pathing with FOV interruption.
|
||||
|
||||
Action Categories:
|
||||
- FREE: LOOK, SPEAK, ANNOUNCE (don't end turn)
|
||||
- FULL: MOVE, WAIT (end turn)
|
||||
|
||||
Constraints:
|
||||
- Only ONE speech action per turn
|
||||
- LOOK provides description and prompts for another action
|
||||
- Multi-tile paths continue without LLM until FOV changes
|
||||
"""
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import List, Tuple, Optional, Set, Dict, Any
|
||||
from enum import Enum
|
||||
|
||||
from action_parser import Action, ActionType
|
||||
|
||||
|
||||
class TurnCost(Enum):
|
||||
"""How much of a turn an action consumes."""
|
||||
FREE = "free" # Doesn't end turn
|
||||
FULL = "full" # Ends turn
|
||||
|
||||
|
||||
# Action cost mapping
|
||||
ACTION_COSTS = {
|
||||
ActionType.LOOK: TurnCost.FREE,
|
||||
ActionType.SPEAK: TurnCost.FREE,
|
||||
ActionType.ANNOUNCE: TurnCost.FREE,
|
||||
ActionType.GO: TurnCost.FULL,
|
||||
ActionType.WAIT: TurnCost.FULL,
|
||||
ActionType.TAKE: TurnCost.FULL,
|
||||
ActionType.DROP: TurnCost.FULL,
|
||||
ActionType.PUSH: TurnCost.FULL,
|
||||
ActionType.USE: TurnCost.FULL,
|
||||
ActionType.OPEN: TurnCost.FULL,
|
||||
ActionType.CLOSE: TurnCost.FULL,
|
||||
ActionType.INVALID: TurnCost.FULL, # Invalid action ends turn
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class TurnState:
|
||||
"""
|
||||
Tracks state within a single turn.
|
||||
|
||||
Used to enforce constraints like "only one speech per turn"
|
||||
and track free actions taken before turn-ending action.
|
||||
"""
|
||||
has_spoken: bool = False
|
||||
free_actions: List[Dict[str, Any]] = field(default_factory=list)
|
||||
turn_ended: bool = False
|
||||
|
||||
def can_speak(self) -> bool:
|
||||
"""Check if agent can still speak this turn."""
|
||||
return not self.has_spoken
|
||||
|
||||
def record_speech(self):
|
||||
"""Record that agent has spoken this turn."""
|
||||
self.has_spoken = True
|
||||
|
||||
def record_free_action(self, action_type: str, details: Dict[str, Any]):
|
||||
"""Record a free action for logging."""
|
||||
self.free_actions.append({
|
||||
"type": action_type,
|
||||
**details
|
||||
})
|
||||
|
||||
def end_turn(self):
|
||||
"""Mark turn as ended."""
|
||||
self.turn_ended = True
|
||||
|
||||
|
||||
@dataclass
|
||||
class PathState:
|
||||
"""
|
||||
Tracks multi-tile movement path for an agent.
|
||||
|
||||
When an agent decides to move to a distant location,
|
||||
we store the path and continue moving without LLM calls
|
||||
until the path completes or FOV changes.
|
||||
"""
|
||||
path: List[Tuple[int, int]] = field(default_factory=list)
|
||||
current_index: int = 0
|
||||
destination_description: str = "" # "the armory", "the door"
|
||||
|
||||
# FOV state when path was planned
|
||||
visible_entities_at_start: Set[str] = field(default_factory=set)
|
||||
|
||||
@property
|
||||
def has_path(self) -> bool:
|
||||
"""Check if there's an active path."""
|
||||
return len(self.path) > self.current_index
|
||||
|
||||
@property
|
||||
def next_tile(self) -> Optional[Tuple[int, int]]:
|
||||
"""Get next tile in path, or None if path complete."""
|
||||
if self.has_path:
|
||||
return self.path[self.current_index]
|
||||
return None
|
||||
|
||||
@property
|
||||
def remaining_tiles(self) -> int:
|
||||
"""Number of tiles left in path."""
|
||||
return max(0, len(self.path) - self.current_index)
|
||||
|
||||
def advance(self):
|
||||
"""Move to next tile in path."""
|
||||
if self.has_path:
|
||||
self.current_index += 1
|
||||
|
||||
def clear(self):
|
||||
"""Clear the current path."""
|
||||
self.path = []
|
||||
self.current_index = 0
|
||||
self.destination_description = ""
|
||||
self.visible_entities_at_start = set()
|
||||
|
||||
def should_interrupt(self, current_visible_entities: Set[str]) -> bool:
|
||||
"""
|
||||
Check if path should be interrupted due to FOV change.
|
||||
|
||||
Returns True if a NEW entity has entered the agent's FOV
|
||||
since the path was planned.
|
||||
"""
|
||||
new_entities = current_visible_entities - self.visible_entities_at_start
|
||||
return len(new_entities) > 0
|
||||
|
||||
|
||||
@dataclass
|
||||
class PointOfInterest:
|
||||
"""
|
||||
A targetable object/location for LOOK/MOVE actions.
|
||||
|
||||
Listed in LLM prompts to guide valid targeting.
|
||||
"""
|
||||
name: str # Short name: "door", "rat", "button"
|
||||
display_name: str # Full description: "a wooden door to the east"
|
||||
position: Tuple[int, int] # Tile coordinates
|
||||
direction: str # Cardinal direction from agent: "north", "east"
|
||||
distance: int # Manhattan distance from agent
|
||||
can_look: bool = True # Can be examined with LOOK
|
||||
can_move_to: bool = False # Can be targeted with GO TO
|
||||
entity_id: Optional[str] = None # Entity ID if this is an entity
|
||||
|
||||
|
||||
def get_action_cost(action: Action) -> TurnCost:
|
||||
"""Get the turn cost for an action."""
|
||||
return ACTION_COSTS.get(action.type, TurnCost.FULL)
|
||||
|
||||
|
||||
def get_direction_name(from_pos: Tuple[int, int], to_pos: Tuple[int, int]) -> str:
|
||||
"""Get cardinal direction name from one position to another."""
|
||||
dx = to_pos[0] - from_pos[0]
|
||||
dy = to_pos[1] - from_pos[1]
|
||||
|
||||
if abs(dx) > abs(dy):
|
||||
return "east" if dx > 0 else "west"
|
||||
elif abs(dy) > abs(dx):
|
||||
return "south" if dy > 0 else "north"
|
||||
else:
|
||||
# Diagonal
|
||||
ns = "south" if dy > 0 else "north"
|
||||
ew = "east" if dx > 0 else "west"
|
||||
return f"{ns}-{ew}"
|
||||
|
||||
|
||||
def manhattan_distance(a: Tuple[int, int], b: Tuple[int, int]) -> int:
|
||||
"""Calculate Manhattan distance between two points."""
|
||||
return abs(a[0] - b[0]) + abs(a[1] - b[1])
|
||||
|
||||
|
||||
class PointOfInterestCollector:
|
||||
"""
|
||||
Collects points of interest visible to an agent.
|
||||
|
||||
Used to populate LLM prompts with valid LOOK/MOVE targets.
|
||||
"""
|
||||
|
||||
def __init__(self, grid, agent_pos: Tuple[int, int]):
|
||||
self.grid = grid
|
||||
self.agent_pos = agent_pos
|
||||
self.points: List[PointOfInterest] = []
|
||||
|
||||
def collect_from_fov(self, world_graph=None) -> List[PointOfInterest]:
|
||||
"""
|
||||
Collect all points of interest visible in current FOV.
|
||||
|
||||
Examines:
|
||||
- Entities (other agents, NPCs, items)
|
||||
- Doors/exits
|
||||
- Interactive objects (buttons, chests)
|
||||
- Notable tiles (walls with features)
|
||||
"""
|
||||
self.points = []
|
||||
|
||||
# Collect entities
|
||||
for entity in self.grid.entities:
|
||||
ex, ey = int(entity.pos[0]), int(entity.pos[1])
|
||||
if (ex, ey) == self.agent_pos:
|
||||
continue # Skip self
|
||||
|
||||
if self.grid.is_in_fov(ex, ey):
|
||||
direction = get_direction_name(self.agent_pos, (ex, ey))
|
||||
distance = manhattan_distance(self.agent_pos, (ex, ey))
|
||||
|
||||
# Try to get entity name/description
|
||||
entity_name = getattr(entity, 'name', None) or f"creature"
|
||||
entity_id = getattr(entity, 'id', None) or str(id(entity))
|
||||
|
||||
self.points.append(PointOfInterest(
|
||||
name=entity_name,
|
||||
display_name=f"a {entity_name} to the {direction}",
|
||||
position=(ex, ey),
|
||||
direction=direction,
|
||||
distance=distance,
|
||||
can_look=True,
|
||||
can_move_to=False, # Can't move onto entities
|
||||
entity_id=entity_id
|
||||
))
|
||||
|
||||
# Collect from WorldGraph if provided
|
||||
if world_graph:
|
||||
self._collect_from_world_graph(world_graph)
|
||||
|
||||
# Sort by distance
|
||||
self.points.sort(key=lambda p: p.distance)
|
||||
|
||||
return self.points
|
||||
|
||||
def _collect_from_world_graph(self, world):
|
||||
"""Collect doors and objects from WorldGraph."""
|
||||
agent_room = world.room_at(*self.agent_pos)
|
||||
if not agent_room:
|
||||
return
|
||||
|
||||
# Doors
|
||||
for door in world.get_exits(agent_room.name):
|
||||
dx, dy = door.position
|
||||
if self.grid.is_in_fov(dx, dy):
|
||||
direction = get_direction_name(self.agent_pos, (dx, dy))
|
||||
distance = manhattan_distance(self.agent_pos, (dx, dy))
|
||||
|
||||
# Get destination room name
|
||||
if door.room_a == agent_room.name:
|
||||
dest = world.rooms.get(door.room_b)
|
||||
else:
|
||||
dest = world.rooms.get(door.room_a)
|
||||
dest_name = dest.display_name if dest else "unknown"
|
||||
|
||||
lock_str = " (locked)" if door.locked else ""
|
||||
|
||||
self.points.append(PointOfInterest(
|
||||
name="door",
|
||||
display_name=f"a door to {dest_name}{lock_str} ({direction})",
|
||||
position=(dx, dy),
|
||||
direction=direction,
|
||||
distance=distance,
|
||||
can_look=True,
|
||||
can_move_to=not door.locked
|
||||
))
|
||||
|
||||
# Objects in room
|
||||
for obj in world.get_objects_in_room(agent_room.name):
|
||||
ox, oy = obj.position
|
||||
if self.grid.is_in_fov(ox, oy):
|
||||
direction = get_direction_name(self.agent_pos, (ox, oy))
|
||||
distance = manhattan_distance(self.agent_pos, (ox, oy))
|
||||
|
||||
self.points.append(PointOfInterest(
|
||||
name=obj.name,
|
||||
display_name=f"{obj.display_name} ({direction})",
|
||||
position=(ox, oy),
|
||||
direction=direction,
|
||||
distance=distance,
|
||||
can_look=True,
|
||||
can_move_to="pressable" not in obj.affordances # Can walk to items
|
||||
))
|
||||
|
||||
def format_for_prompt(self) -> str:
|
||||
"""Format points of interest for inclusion in LLM prompt."""
|
||||
if not self.points:
|
||||
return "No notable objects in view."
|
||||
|
||||
lines = ["Points of interest:"]
|
||||
for poi in self.points:
|
||||
actions = []
|
||||
if poi.can_look:
|
||||
actions.append(f"LOOK AT {poi.name.upper()}")
|
||||
if poi.can_move_to:
|
||||
actions.append(f"GO TO {poi.name.upper()}")
|
||||
|
||||
action_str = ", ".join(actions) if actions else "observe only"
|
||||
lines.append(f" - {poi.display_name}: {action_str}")
|
||||
|
||||
return "\n".join(lines)
|
||||
Loading…
Add table
Add a link
Reference in a new issue