mirror of
https://github.com/HKUDS/OpenSpace.git
synced 2026-09-11 22:51:05 +00:00
271 lines
No EOL
8.9 KiB
Python
271 lines
No EOL
8.9 KiB
Python
"""
|
|
Agent Action Recorder
|
|
|
|
Records agent decision-making processes, reasoning, and outputs.
|
|
Focuses on high-level agent behaviors rather than low-level tool executions.
|
|
"""
|
|
|
|
import datetime
|
|
import json
|
|
from typing import Any, Dict, Optional
|
|
from pathlib import Path
|
|
|
|
from openspace.utils.logging import Logger
|
|
|
|
logger = Logger.get_logger(__name__)
|
|
|
|
|
|
class ActionRecorder:
|
|
"""
|
|
Records agent actions and decision-making processes.
|
|
|
|
This recorder captures the 'thinking' layer of the agent:
|
|
- Task planning and decomposition
|
|
- Tool selection reasoning
|
|
- Evaluation decisions
|
|
"""
|
|
|
|
def __init__(self, trajectory_dir: Path):
|
|
"""
|
|
Initialize action recorder.
|
|
|
|
Args:
|
|
trajectory_dir: Directory to save action records
|
|
"""
|
|
self.trajectory_dir = trajectory_dir
|
|
self.actions_file = trajectory_dir / "agent_actions.jsonl"
|
|
self.step_counter = 0
|
|
|
|
# Ensure directory exists
|
|
self.trajectory_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
async def record_action(
|
|
self,
|
|
agent_name: str,
|
|
action_type: str,
|
|
input_data: Optional[Dict[str, Any]] = None,
|
|
reasoning: Optional[Dict[str, Any]] = None,
|
|
output_data: Optional[Dict[str, Any]] = None,
|
|
metadata: Optional[Dict[str, Any]] = None,
|
|
related_tool_steps: Optional[list] = None,
|
|
correlation_id: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Record an agent action.
|
|
|
|
Args:
|
|
agent_name: Name of the agent performing the action
|
|
action_type: Type of action (plan | execute | evaluate | monitor)
|
|
input_data: Input data the agent received (simplified)
|
|
reasoning: Agent's reasoning process (structured)
|
|
output_data: Agent's output/decision (structured)
|
|
metadata: Additional metadata (LLM model, tokens, duration, etc.)
|
|
related_tool_steps: List of tool execution step numbers related to this action
|
|
correlation_id: Optional correlation ID to link related events
|
|
"""
|
|
self.step_counter += 1
|
|
timestamp = datetime.datetime.now().isoformat()
|
|
|
|
# Infer agent type from agent name
|
|
agent_type = self._infer_agent_type(agent_name)
|
|
|
|
action_info = {
|
|
"step": self.step_counter,
|
|
"timestamp": timestamp,
|
|
"agent_name": agent_name,
|
|
"agent_type": agent_type,
|
|
"action_type": action_type,
|
|
"correlation_id": correlation_id or f"action_{self.step_counter}_{timestamp}",
|
|
}
|
|
|
|
# Add input (with smart truncation)
|
|
if input_data:
|
|
action_info["input"] = self._truncate_data(input_data, max_length=1000)
|
|
|
|
# Add reasoning (keep structured)
|
|
if reasoning:
|
|
action_info["reasoning"] = self._truncate_data(reasoning, max_length=2000)
|
|
|
|
# Add output (keep structured)
|
|
if output_data:
|
|
action_info["output"] = self._truncate_data(output_data, max_length=1000)
|
|
|
|
# Add metadata
|
|
if metadata:
|
|
action_info["metadata"] = metadata
|
|
|
|
# Add related tool steps for correlation
|
|
if related_tool_steps:
|
|
action_info["related_tool_steps"] = related_tool_steps
|
|
|
|
# Append to JSONL file
|
|
await self._append_to_file(action_info)
|
|
|
|
logger.debug(
|
|
f"Recorded {action_type} action from {agent_name} (step {self.step_counter})"
|
|
)
|
|
|
|
return action_info
|
|
|
|
def _infer_agent_type(self, agent_name: str) -> str:
|
|
name_lower = agent_name.lower()
|
|
|
|
if "host" in name_lower:
|
|
return "host"
|
|
elif "grounding" in name_lower:
|
|
return "grounding"
|
|
elif "eval" in name_lower:
|
|
return "eval"
|
|
elif "coordinator" in name_lower:
|
|
return "coordinator"
|
|
else:
|
|
return "unknown"
|
|
|
|
def _truncate_data(self, data: Any, max_length: int) -> Any:
|
|
if isinstance(data, str):
|
|
if len(data) > max_length:
|
|
return data[:max_length] + "... [truncated]"
|
|
return data
|
|
|
|
elif isinstance(data, dict):
|
|
result = {}
|
|
for key, value in data.items():
|
|
if isinstance(value, str) and len(value) > max_length:
|
|
result[key] = value[:max_length] + "... [truncated]"
|
|
elif isinstance(value, (dict, list)):
|
|
# Recursively truncate nested structures
|
|
result[key] = self._truncate_data(value, max_length)
|
|
else:
|
|
result[key] = value
|
|
return result
|
|
|
|
elif isinstance(data, list):
|
|
# Truncate list items
|
|
result = []
|
|
for item in data:
|
|
if isinstance(item, str) and len(item) > max_length:
|
|
result.append(item[:max_length] + "... [truncated]")
|
|
elif isinstance(item, (dict, list)):
|
|
result.append(self._truncate_data(item, max_length))
|
|
else:
|
|
result.append(item)
|
|
return result
|
|
|
|
else:
|
|
return data
|
|
|
|
async def _append_to_file(self, action_info: Dict[str, Any]):
|
|
"""Append action to JSONL file."""
|
|
with open(self.actions_file, "a", encoding="utf-8") as f:
|
|
f.write(json.dumps(action_info, ensure_ascii=False))
|
|
f.write("\n")
|
|
|
|
def get_step_count(self) -> int:
|
|
"""Get current step count."""
|
|
return self.step_counter
|
|
|
|
|
|
def load_agent_actions(trajectory_dir: str) -> list:
|
|
"""
|
|
Load agent actions from a trajectory directory.
|
|
"""
|
|
actions_file = Path(trajectory_dir) / "agent_actions.jsonl"
|
|
|
|
if not actions_file.exists():
|
|
logger.warning(f"Agent actions file not found: {actions_file}")
|
|
return []
|
|
|
|
actions = []
|
|
try:
|
|
with open(actions_file, "r", encoding="utf-8") as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if line:
|
|
actions.append(json.loads(line))
|
|
|
|
logger.info(f"Loaded {len(actions)} agent actions from {actions_file}")
|
|
return actions
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to load agent actions from {actions_file}: {e}")
|
|
return []
|
|
|
|
|
|
def analyze_agent_actions(actions: list) -> Dict[str, Any]:
|
|
"""
|
|
Analyze agent actions and generate statistics.
|
|
"""
|
|
if not actions:
|
|
return {
|
|
"total_actions": 0,
|
|
"by_agent": {},
|
|
"by_type": {},
|
|
}
|
|
|
|
# Count by agent
|
|
by_agent = {}
|
|
by_type = {}
|
|
|
|
for action in actions:
|
|
agent_name = action.get("agent_name", "unknown")
|
|
action_type = action.get("action_type", "unknown")
|
|
|
|
by_agent[agent_name] = by_agent.get(agent_name, 0) + 1
|
|
by_type[action_type] = by_type.get(action_type, 0) + 1
|
|
|
|
return {
|
|
"total_actions": len(actions),
|
|
"by_agent": by_agent,
|
|
"by_type": by_type,
|
|
}
|
|
|
|
|
|
def format_agent_actions(actions: list, format_type: str = "compact") -> str:
|
|
"""
|
|
Format agent actions for display.
|
|
"""
|
|
if not actions:
|
|
return "No agent actions recorded"
|
|
|
|
if format_type == "compact":
|
|
lines = []
|
|
for action in actions:
|
|
step = action.get("step", "?")
|
|
agent = action.get("agent_name", "?")
|
|
action_type = action.get("action_type", "?")
|
|
|
|
# Try to extract key info from reasoning or output
|
|
key_info = ""
|
|
if action.get("reasoning"):
|
|
thought = action["reasoning"].get("thought", "")
|
|
if thought:
|
|
key_info = f": {thought[:60]}..."
|
|
|
|
lines.append(f"Step {step}: [{agent}] {action_type}{key_info}")
|
|
|
|
return "\n".join(lines)
|
|
|
|
elif format_type == "detailed":
|
|
lines = []
|
|
for action in actions:
|
|
lines.append(f"\n{'='*60}")
|
|
lines.append(f"Step {action.get('step', '?')}: {action.get('agent_name', '?')}")
|
|
lines.append(f"Type: {action.get('action_type', '?')}")
|
|
lines.append(f"Time: {action.get('timestamp', '?')}")
|
|
|
|
if action.get("reasoning"):
|
|
lines.append("\nReasoning:")
|
|
lines.append(json.dumps(action["reasoning"], indent=2, ensure_ascii=False))
|
|
|
|
if action.get("output"):
|
|
lines.append("\nOutput:")
|
|
lines.append(json.dumps(action["output"], indent=2, ensure_ascii=False))
|
|
|
|
if action.get("metadata"):
|
|
lines.append("\nMetadata:")
|
|
lines.append(json.dumps(action["metadata"], indent=2, ensure_ascii=False))
|
|
|
|
return "\n".join(lines)
|
|
|
|
else:
|
|
raise ValueError(f"Unknown format type: {format_type}") |