system-hint

第2章 · 上下文工程 · 配套项目 chapter2/system-hint

项目说明

System-Hint Enhanced AI Agent

对应书中 实验 2-8：几种好用的 Agent 状态栏技术（第二章“Agent 状态栏 / Agent Status Bar”一节）。本目录即书中所说的 agent-status-bar 实验框架——“system hint（系统提示）”与“Agent 状态栏（status bar）”是同一概念的两种叫法：在上下文末尾以一条 role=user 的消息注入动态状态摘要。

An advanced AI agent that demonstrates the power of system hints for improving agent trajectory and preventing common issues like infinite loops, poor context awareness, and inefficient task management, with automatic trajectory saving for debugging.

⚡ 先跑离线预览（无需 API Key）

想在不配置任何 API Key 的情况下直观看到状态栏如何改变模型看到的上下文，运行：

python main.py --mode preview

该命令在本地渲染书中五种状态栏技术（时间戳、工具调用计数器、TODO 列表、详细错误信息、系统状态感知）， 对每一项做一次 “无状态栏 vs 有状态栏” 的前后对比，并打印最终追加到上下文末尾的完整状态栏消息。 配合 --no-timestamps / --no-counter / --no-todo / --no-errors / --no-state 可分别关闭某一类， 单独观察它对上下文的影响。整个过程不发起任何 LLM 调用。

🌟 Key Features

1. Timestamp Tracking

• Adds timestamps to user messages and tool call results
• Helps the agent understand temporal context
• Simulates time delays for realistic multi-day scenarios

2. Tool Call Counter

• Tracks how many times each tool has been called
• Prevents infinite loops and repetitive behavior
• Shows call number in tool responses (e.g., "Tool call #3 for 'read_file'")

3. TODO List Management

• Built-in task tracking system with management rules in system prompt
• Four states: pending, in_progress, completed, cancelled
• Persistent across conversation with rewrite and update capabilities
• Agent automatically creates TODO lists for complex tasks (3+ steps)
• Helps maintain focus and track progress systematically

4. Detailed Error Messages

• Provides comprehensive error information instead of generic messages
• Includes error type, arguments, traceback (in verbose mode)
• Offers suggestions for fixing common errors
• Helps agent learn from failures and adapt strategies

5. System State Awareness

• Tracks current directory, shell environment, and system information
• Updates dynamically as agent navigates filesystem
• Provides context for command execution

6. Automatic Trajectory Saving

• Saves full conversation history and state to trajectory.json after each iteration
• Captures complete debugging information even if execution fails
• Includes conversation history, tool calls, TODO lists, and configuration
• Provides view_trajectory.py utility for analyzing saved trajectories

🚀 Quick Start

Installation

# Clone the repository (if not already done)
cd chapter2/system-hint

# Install dependencies
pip install -r requirements.txt

# Copy and configure environment
cp env.example .env
# Edit .env with your KIMI_API_KEY
export KIMI_API_KEY='your-api-key-here'

通用回退（OpenRouter）：未设置 KIMI_API_KEY 时，只要配置了 OPENROUTER_API_KEY，实验会自动改走 OpenRouter（kimi-* 会映射为 moonshotai/kimi-k2）。设置了 KIMI_API_KEY 时行为完全不变。

Basic Usage

# Offline preview of the status bar (no API key required)
python main.py --mode preview

# Interactive mode (default)
python main.py

# Run the sample task (analyze week1/week2 projects)
python main.py --mode sample

# Execute a single task from command line
python main.py --mode single --task "Create a hello world Python script"

# Override provider / model, and choose the trajectory output file
python main.py --mode single --task "..." --provider kimi --model kimi-k3 --output run1.json

# Run demonstrations
python main.py --mode demo --demo basic
python main.py --mode demo --demo loop
python main.py --mode demo --demo comparison

# Disable specific features (works for both preview and live modes)
python main.py --mode single --no-todo --no-timestamps --task "Simple task"
# Or observe the effect offline (no API key):
python main.py --mode preview --no-todo --no-timestamps

# Quick start with sample task
python quickstart.py

# View saved trajectory after running any task
python view_trajectory.py

# View a specific trajectory file
python view_trajectory.py path/to/trajectory.json

Programmatic Usage

from agent import SystemHintAgent, SystemHintConfig

# Configure system hints
config = SystemHintConfig(
    enable_timestamps=True,
    enable_tool_counter=True,
    enable_todo_list=True,
    enable_detailed_errors=True,
    enable_system_state=True,
    save_trajectory=True,
    trajectory_file="my_trajectory.json"
)

# Create agent
agent = SystemHintAgent(
    api_key="your-api-key",
    provider="kimi",
    config=config,
    verbose=False
)

# Execute task
task = "Create a Python script that analyzes CSV files"
result = agent.execute_task(task, max_iterations=20)

print(f"Success: {result['success']}")
print(f"Final answer: {result['final_answer']}")
print(f"Trajectory saved to: {result['trajectory_file']}")

📂 Project Structure

system-hint/
├── agent.py          # Main agent implementation with system hints
├── main.py           # CLI interface with multiple modes
├── config.py         # Configuration management
├── quickstart.py     # Quick demo script
├── test_basic.py     # Basic tests
├── test_hint_behavior.py # System-hint behavior tests
├── view_trajectory.py # Trajectory viewing utility
├── requirements.txt  # Python dependencies
├── env.example       # Environment variable template
├── trajectory.json   # Auto-saved trajectory (created at runtime)
├── CHANGELOG.md      # Change log
├── NOTES.md          # Design notes
└── README.md        # This file

🔬 How System Hints Work

System Hint Architecture

System hints are contextual information added to the conversation as temporary user messages before sending to the LLM. They are NOT stored in the conversation history, preventing context pollution while providing crucial state information.

# System hint example (added as user message before LLM call):
=== SYSTEM STATE ===
Current Time: 2024-12-13 10:30:45
Current Directory: /home/user/projects
System: Linux (5.15.0)
Shell Environment: Linux Shell (bash)
Python Version: 3.10.0

=== CURRENT TASKS ===
TODO List:
  [1] 🔄 Read configuration file (in_progress)
  [2] ⏳ Process data (pending)
  [3] ✅ Initialize environment (completed)

Management Rules in System Prompt

The system prompt includes explicit rules for TODO management, error handling, and tool usage patterns that guide the agent's behavior:

• Automatic TODO list creation for complex tasks
• Status management (only one in_progress at a time)
• Tool call awareness (check call numbers)
• Error recovery strategies

🛠 Configuration Options

SystemHintConfig Parameters

Parameter	Default	Description
enable_timestamps	True	Add timestamps to messages
enable_tool_counter	True	Track tool call counts
enable_todo_list	True	Enable TODO list management
enable_detailed_errors	True	Provide detailed error information
enable_system_state	True	Track system state
timestamp_format	"%Y-%m-%d %H:%M:%S"	Timestamp format string
simulate_time_delay	False	Simulate time passing (demo)
save_trajectory	True	Save trajectory to file
trajectory_file	"trajectory.json"	Trajectory output file

🔍 Demonstrations

0. Offline Status-Bar Preview (no API key)

Renders the five status-bar techniques as before/after comparisons, entirely offline:

python main.py --mode preview

1. Basic Features Demo

Shows all system hints working together:

python main.py --mode demo --demo basic

2. Loop Prevention Demo

Demonstrates how tool call counting prevents infinite loops:

python main.py --mode demo --demo loop

3. Comparison Demo

Shows the difference between with and without system hints:

python main.py --mode demo --demo comparison

🎯 Sample Tasks

The agent includes several pre-configured sample tasks:

1. Project Analysis - Analyze week1/week2 AI agent projects
2. File Operations - Create, read, and modify files
3. Code Generation - Generate Python scripts with specific functionality
4. System Commands - Execute shell commands and process results

📊 Analyzing Results

Viewing Trajectories

After running any task, use the trajectory viewer:

python view_trajectory.py

# Output shows:
# - Total iterations
# - Tool usage statistics
# - TODO list progress
# - Conversation highlights
# - Configuration used

Performance Metrics

The agent tracks: - Number of iterations needed - Tool calls made (with success/failure rates) - TODO completion status - Time taken (when timestamps enabled) - Final success/failure state

🧪 Testing

# Run basic tests
python test_basic.py

# Test specific configurations
python -c "
from agent import SystemHintAgent, SystemHintConfig
config = SystemHintConfig(enable_todo_list=False)
agent = SystemHintAgent('api_key', config=config)
# Test without TODO lists
"

🔧 Troubleshooting

Common Issues

1. API Key Not Set

   export KIMI_API_KEY='your-api-key-here'
   

2. Tool Call Loops

3. Enable tool counter: enable_tool_counter=True

4. Check tool call numbers in output

5. Lost Context

6. Enable timestamps: enable_timestamps=True

7. Enable system state: enable_system_state=True

8. Task Management Issues

9. Enable TODO list: enable_todo_list=True
10. Agent will automatically organize complex tasks

📝 Notes

• System hints are added as temporary user messages, not stored in history
• Trajectory files capture complete execution state for debugging
• TODO lists help agents maintain focus on multi-step tasks
• Tool call counters prevent infinite loops automatically
• Detailed errors with suggestions help agents self-correct

🤝 Contributing

Feel free to enhance the system hint features: - Add new hint types - Improve error suggestions - Enhance TODO management rules - Add visualization tools for trajectories

📄 License

MIT License - See LICENSE file for details

源代码

agent.py

"""
System-Hint Enhanced AI Agent
An agent that demonstrates advanced trajectory management with system hints,
including timestamps, tool call tracking, TODO lists, and detailed error messages.
"""

import json
import os
import sys
import subprocess
import platform
import logging
from typing import List, Dict, Any, Optional, Tuple
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime, timedelta
import requests
from openai import OpenAI
import traceback
import tempfile
import shutil
from pathlib import Path


def _reasoning_safe_temperature(model, requested=1.0):
    """Reasoning models (Kimi K3, GPT-5, ...) only accept temperature=1.
    Return 1 for those; otherwise the requested value so non-reasoning
    providers (Doubao, DeepSeek, older Moonshot) are unchanged."""
    m = str(model or "").lower().replace("/", "-")
    return 1 if ("kimi-k3" in m or "gpt-5" in m) else requested

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)


class TodoStatus(Enum):
    """Status of a TODO item"""
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    CANCELLED = "cancelled"


@dataclass
class TodoItem:
    """Represents a single TODO item"""
    id: int
    content: str
    status: TodoStatus = TodoStatus.PENDING
    created_at: str = field(default_factory=lambda: datetime.now().isoformat())
    updated_at: Optional[str] = None


@dataclass
class ToolCall:
    """Represents a single tool call with enhanced tracking"""
    tool_name: str
    arguments: Dict[str, Any]
    result: Optional[Any] = None
    error: Optional[str] = None
    call_number: int = 1  # Track how many times this tool has been called
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
    duration_ms: Optional[int] = None


@dataclass
class SystemHintConfig:
    """Configuration for system hints"""
    enable_timestamps: bool = True
    enable_tool_counter: bool = True
    enable_todo_list: bool = True
    enable_detailed_errors: bool = True
    enable_system_state: bool = True  # Current dir, shell, etc.
    timestamp_format: str = "%Y-%m-%d %H:%M:%S"
    simulate_time_delay: bool = False  # For demo purposes
    save_trajectory: bool = True  # Save conversation history to file
    trajectory_file: str = "trajectory.json"  # File to save trajectory to


class SystemHintAgent:
    """
    AI Agent with enhanced system hints for better trajectory management
    """

    def __init__(self, api_key: str, provider: str = "kimi", 
                 model: Optional[str] = None, config: Optional[SystemHintConfig] = None,
                 verbose: bool = True):
        """
        Initialize the agent

        Args:
            api_key: API key for the LLM provider
            provider: LLM provider ('kimi' for Kimi K3)
            model: Optional model override
            config: System hint configuration
            verbose: If True, log full details
        """
        self.provider = provider.lower()
        self.verbose = verbose
        self.config = config or SystemHintConfig()

        # Configure client based on provider
        if self.provider == "kimi" or self.provider == "moonshot":
            # 默认 Moonshot/Kimi 官方端点；若传入 OpenRouter key（sk-or-…）则自动
            # 回退到 OpenRouter，并把 kimi-* 映射为 moonshotai/kimi-k2。
            from openrouter_fallback import (
                OPENROUTER_BASE_URL,
                is_openrouter_key,
                map_model_to_openrouter,
            )
            resolved_model = model or "kimi-k3"
            if is_openrouter_key(api_key):
                base_url = OPENROUTER_BASE_URL
                resolved_model = map_model_to_openrouter(resolved_model)
            else:
                base_url = "https://api.moonshot.cn/v1"
            self.client = OpenAI(
                api_key=api_key,
                base_url=base_url
            )
            self.model = resolved_model
        else:
            raise ValueError(f"Unsupported provider: {provider}")

        # Initialize tracking
        self.tool_call_counts: Dict[str, int] = {}
        self.tool_calls: List[ToolCall] = []
        self.todo_list: List[TodoItem] = []
        self.next_todo_id = 1

        # Initialize conversation history
        self.conversation_history = []
        self.simulated_time = datetime.now()  # For demo time simulation
        self._init_system_prompt()

        # Track current working directory
        self.current_directory = os.getcwd()

        # Track last messages sent to LLM
        self.last_llm_messages = None

        logger.info(f"System-Hint Agent initialized with provider: {self.provider}, model: {self.model}")

    def _init_system_prompt(self):
        """Initialize the system prompt for the conversation"""
        system_content = """You are an intelligent assistant with access to various tools for file operations, code execution, and system commands.

Your task is to complete the given objectives efficiently using the available tools. Think step by step and use tools as needed.

## TODO List Management Rules:
- For any complex task with 3+ distinct steps, immediately create a TODO list using `rewrite_todo_list`
- Break down the user's request into specific, actionable TODO items
- Update TODO items to 'in_progress' when starting work on them using `update_todo_status`
- Mark items as 'completed' immediately after finishing them
- Only have ONE item 'in_progress' at a time
- If you encounter errors or need to change approach, update relevant TODOs to 'cancelled' and add new ones
- Use the TODO list as your primary planning and tracking mechanism
- Reference TODO items by their ID when discussing progress

## Key Behaviors:
1. ALWAYS start complex tasks by creating a TODO list
2. Pay attention to timestamps to understand the timeline of events
3. Notice tool call numbers (e.g., "Tool call #3") to avoid repetitive loops - if you see high numbers, change strategy
4. Learn from detailed error messages to fix issues and adapt your approach
5. Be aware of your current directory and system environment shown in system state
6. When exploring projects, systematically read key files (README, main.py, agent.py) to understand structure

## Error Handling:
- Read error messages carefully - they contain specific information about what went wrong
- Use the suggestions provided in error messages to fix issues
- If a tool fails multiple times (check the call number), try a different approach
- Common fixes: check file paths, verify current directory, ensure proper permissions

Important: When you have completed all tasks, clearly state "FINAL ANSWER:" followed by a comprehensive summary of what was accomplished."""

        self.conversation_history = [
            {
                "role": "system",
                "content": system_content
            }
        ]

    def _get_system_state(self) -> str:
        """Get current system state information"""
        if not self.config.enable_system_state:
            return ""

        # Detect OS
        system = platform.system()
        if system == "Windows":
            shell_type = "Windows Command Prompt or PowerShell"
        elif system == "Darwin":
            shell_type = "macOS Terminal (zsh/bash)"
        else:
            shell_type = f"Linux Shell ({os.environ.get('SHELL', 'bash')})"

        state_info = [
            f"Current Time: {self._get_timestamp()}",
            f"Current Directory: {self.current_directory}",
            f"System: {system} ({platform.release()})",
            f"Shell Environment: {shell_type}",
            f"Python Version: {sys.version.split()[0]}"
        ]

        return "\n".join(state_info)

    def _get_timestamp(self) -> str:
        """Get formatted timestamp"""
        if self.config.simulate_time_delay:
            # For demo: simulate time passing
            return self.simulated_time.strftime(self.config.timestamp_format)
        return datetime.now().strftime(self.config.timestamp_format)

    def _advance_simulated_time(self, hours: int = 0, minutes: int = 0, seconds: int = 30):
        """Advance simulated time for demo purposes"""
        if self.config.simulate_time_delay:
            self.simulated_time += timedelta(hours=hours, minutes=minutes, seconds=seconds)

    def _save_trajectory(self, iteration: int, final_answer: Optional[str] = None):
        """Save current trajectory to JSON file for debugging"""
        if not self.config.save_trajectory:
            return

        trajectory_data = {
            "timestamp": datetime.now().isoformat(),
            "iteration": iteration,
            "provider": self.provider,
            "model": self.model,
            "conversation_history": self.conversation_history,
            "last_llm_messages": self.last_llm_messages,
            "tool_calls": [
                {
                    "tool_name": call.tool_name,
                    "arguments": call.arguments,
                    "result": call.result,
                    "error": call.error,
                    "call_number": call.call_number,
                    "timestamp": call.timestamp,
                    "duration_ms": call.duration_ms
                }
                for call in self.tool_calls
            ],
            "todo_list": [
                {
                    "id": item.id,
                    "content": item.content,
                    "status": item.status.value,
                    "created_at": item.created_at,
                    "updated_at": item.updated_at
                }
                for item in self.todo_list
            ],
            "current_directory": self.current_directory,
            "final_answer": final_answer,
            "config": {
                "enable_timestamps": self.config.enable_timestamps,
                "enable_tool_counter": self.config.enable_tool_counter,
                "enable_todo_list": self.config.enable_todo_list,
                "enable_detailed_errors": self.config.enable_detailed_errors,
                "enable_system_state": self.config.enable_system_state,
                "timestamp_format": self.config.timestamp_format,
                "simulate_time_delay": self.config.simulate_time_delay
            }
        }

        try:
            # Save to file, overwriting each time to capture latest state
            with open(self.config.trajectory_file, 'w', encoding='utf-8') as f:
                json.dump(trajectory_data, f, indent=2, ensure_ascii=False)

            if self.verbose:
                logger.info(f"Trajectory saved to {self.config.trajectory_file} (iteration {iteration})")
        except Exception as e:
            logger.warning(f"Failed to save trajectory: {e}")

    def _format_todo_list(self) -> str:
        """Format TODO list for display"""
        if not self.todo_list:
            return "TODO List: Empty"

        lines = ["TODO List:"]
        for item in self.todo_list:
            status_symbol = {
                TodoStatus.PENDING: "⏳",
                TodoStatus.IN_PROGRESS: "🔄",
                TodoStatus.COMPLETED: "✅",
                TodoStatus.CANCELLED: "❌"
            }.get(item.status, "❓")

            lines.append(f"  [{item.id}] {status_symbol} {item.content} ({item.status.value})")

        return "\n".join(lines)

    def _get_system_hint(self) -> Optional[str]:
        """Get system hint content with current state"""
        if not any([self.config.enable_system_state, self.config.enable_todo_list]):
            return None

        hint_parts = []

        if self.config.enable_system_state:
            hint_parts.append("=== SYSTEM STATE ===")
            hint_parts.append(self._get_system_state())
            hint_parts.append("")

        if self.config.enable_todo_list and self.todo_list:
            hint_parts.append("=== CURRENT TASKS ===")
            hint_parts.append(self._format_todo_list())
            hint_parts.append("")

        if hint_parts:
            return "\n".join(hint_parts)
        return None

    def _get_tools_description(self) -> List[Dict[str, Any]]:
        """Get tool descriptions for the model"""
        tools = [
            {
                "type": "function",
                "function": {
                    "name": "read_file",
                    "description": "Read the contents of a text file. Returns error for binary files. Supports partial reading for large files.",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "file_path": {
                                "type": "string",
                                "description": "Path to the file to read (absolute or relative to current directory)"
                            },
                            "begin_line": {
                                "type": "integer",
                                "description": "Optional: Line number to start reading from (1-based indexing). E.g., begin_line=10 starts from line 10."
                            },
                            "number_lines": {
                                "type": "integer",
                                "description": "Optional: Number of lines to read from begin_line. E.g., number_lines=50 reads 50 lines."
                            }
                        },
                        "required": ["file_path"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "write_file",
                    "description": "Write content to a file (creates or overwrites)",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "file_path": {
                                "type": "string",
                                "description": "Path to the file to write"
                            },
                            "content": {
                                "type": "string",
                                "description": "Content to write to the file"
                            }
                        },
                        "required": ["file_path", "content"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "code_interpreter",
                    "description": "Execute Python code in a restricted environment",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "code": {
                                "type": "string",
                                "description": "Python code to execute"
                            }
                        },
                        "required": ["code"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "execute_command",
                    "description": "Execute a shell command in the current directory",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "command": {
                                "type": "string",
                                "description": "Shell command to execute"
                            },
                            "working_dir": {
                                "type": "string",
                                "description": "Optional working directory for the command (defaults to current directory)"
                            }
                        },
                        "required": ["command"]
                    }
                }
            }
        ]

        # Add TODO management tools if enabled
        if self.config.enable_todo_list:
            tools.extend([
                {
                    "type": "function",
                    "function": {
                        "name": "rewrite_todo_list",
                        "description": "Rewrite the TODO list with new pending items (keeps completed/cancelled items)",
                        "parameters": {
                            "type": "object",
                            "properties": {
                                "items": {
                                    "type": "array",
                                    "items": {
                                        "type": "string"
                                    },
                                    "description": "List of new TODO items to add as pending"
                                }
                            },
                            "required": ["items"]
                        }
                    }
                },
                {
                    "type": "function",
                    "function": {
                        "name": "update_todo_status",
                        "description": "Update the status of existing TODO items",
                        "parameters": {
                            "type": "object",
                            "properties": {
                                "updates": {
                                    "type": "array",
                                    "items": {
                                        "type": "object",
                                        "properties": {
                                            "id": {
                                                "type": "integer",
                                                "description": "TODO item ID"
                                            },
                                            "status": {
                                                "type": "string",
                                                "enum": ["pending", "in_progress", "completed", "cancelled"],
                                                "description": "New status for the item"
                                            }
                                        },
                                        "required": ["id", "status"]
                                    },
                                    "description": "List of TODO items to update with their new status"
                                }
                            },
                            "required": ["updates"]
                        }
                    }
                }
            ])

        return tools

    def _execute_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Tuple[Any, Optional[str]]:
        """
        Execute a tool and return the result with detailed error information

        Returns:
            Tuple of (result, error_detail)
        """
        start_time = datetime.now()

        try:
            if tool_name == "read_file":
                result = self._tool_read_file(**arguments)
            elif tool_name == "write_file":
                result = self._tool_write_file(**arguments)
            elif tool_name == "code_interpreter":
                result = self._tool_code_interpreter(**arguments)
            elif tool_name == "execute_command":
                result = self._tool_execute_command(**arguments)
            elif tool_name == "rewrite_todo_list":
                result = self._tool_rewrite_todo_list(**arguments)
            elif tool_name == "update_todo_status":
                result = self._tool_update_todo_status(**arguments)
            else:
                error = f"Unknown tool: {tool_name}"
                return {"error": error}, error

            duration_ms = int((datetime.now() - start_time).total_seconds() * 1000)
            return result, None

        except Exception as e:
            duration_ms = int((datetime.now() - start_time).total_seconds() * 1000)

            # Get detailed error information
            error_detail = self._get_detailed_error(e, tool_name, arguments)

            if self.config.enable_detailed_errors:
                return {"error": error_detail}, error_detail
            else:
                return {"error": str(e)}, str(e)

    def _get_detailed_error(self, exception: Exception, tool_name: str, arguments: Dict[str, Any]) -> str:
        """Get detailed error information for debugging"""
        error_parts = [
            f"Tool '{tool_name}' failed with {type(exception).__name__}: {str(exception)}",
            f"Arguments: {json.dumps(arguments, indent=2)}",
        ]

        # Add traceback for debugging
        if self.verbose:
            tb = traceback.format_exc()
            error_parts.append(f"Traceback:\n{tb}")

        # Add suggestions based on error type
        suggestions = self._get_error_suggestions(exception, tool_name)
        if suggestions:
            error_parts.append(f"Suggestions: {suggestions}")

        return "\n".join(error_parts)

    def _get_error_suggestions(self, exception: Exception, tool_name: str) -> str:
        """Get suggestions for fixing common errors"""
        error_str = str(exception).lower()
        exception_type = type(exception).__name__

        suggestions = []

        if "permission" in error_str or exception_type == "PermissionError":
            suggestions.append("Check file/directory permissions")
            suggestions.append("Try using a different directory or running with appropriate permissions")
        elif "not found" in error_str or "no such file" in error_str or exception_type == "FileNotFoundError":
            suggestions.append("Verify the file/directory path exists")
            suggestions.append("Check the current working directory")
            suggestions.append("Use absolute paths or create the file/directory first")
        elif "syntax" in error_str or exception_type == "SyntaxError":
            suggestions.append("Check the code syntax")
            suggestions.append("Ensure proper indentation and valid Python syntax")
        elif "timeout" in error_str:
            suggestions.append("The operation took too long")
            suggestions.append("Try with simpler input or break into smaller steps")
        elif "import" in error_str or exception_type == "ImportError":
            suggestions.append("Required module not available in restricted environment")
            suggestions.append("Use only built-in Python modules")

        return " | ".join(suggestions) if suggestions else ""

    # Tool implementations
    def _tool_read_file(self, file_path: str, begin_line: Optional[int] = None, 
                       number_lines: Optional[int] = None) -> Dict[str, Any]:
        """Read file contents with optional line-based reading"""
        try:
            # Resolve path relative to current directory
            if not os.path.isabs(file_path):
                file_path = os.path.join(self.current_directory, file_path)

            # Check if file exists
            if not os.path.exists(file_path):
                raise FileNotFoundError(f"File not found: {file_path}")

            # Check if it's a binary file
            try:
                with open(file_path, 'rb') as f:
                    # Read first 1024 bytes to check for binary content
                    chunk = f.read(1024)
                    # Check for null bytes (common in binary files)
                    if b'\x00' in chunk:
                        return {
                            "success": False,
                            "error": "Cannot read binary file. This tool only supports text files.",
                            "file_path": file_path,
                            "is_binary": True
                        }
                    # Also check if it's valid UTF-8
                    try:
                        chunk.decode('utf-8')
                    except UnicodeDecodeError:
                        return {
                            "success": False,
                            "error": "File is not a valid text file (encoding error).",
                            "file_path": file_path,
                            "is_binary": True
                        }
            except Exception as e:
                # If we can't read it as binary, probably permission issue
                raise

            # Read the file content
            with open(file_path, 'r', encoding='utf-8') as f:
                if begin_line is not None or number_lines is not None:
                    # Line-based reading
                    all_lines = f.readlines()
                    total_lines = len(all_lines)

                    # Calculate line range
                    start_line = (begin_line - 1) if begin_line is not None else 0
                    if start_line < 0:
                        start_line = 0
                    if start_line >= total_lines:
                        return {
                            "success": False,
                            "error": f"begin_line {begin_line} is beyond file length ({total_lines} lines)",
                            "file_path": file_path,
                            "total_lines": total_lines
                        }

                    if number_lines is not None:
                        end_line = min(start_line + number_lines, total_lines)
                    else:
                        end_line = total_lines

                    # Get the requested lines
                    selected_lines = all_lines[start_line:end_line]
                    content = ''.join(selected_lines)

                    # Get file info
                    stat = os.stat(file_path)

                    return {
                        "success": True,
                        "file_path": file_path,
                        "content": content,
                        "size_bytes": stat.st_size,
                        "total_lines": total_lines,
                        "begin_line": start_line + 1,  # Convert back to 1-based
                        "end_line": end_line,
                        "lines_read": len(selected_lines),
                        "partial_read": True
                    }
                else:
                    # Full file reading
                    content = f.read()

                    # Get file info
                    stat = os.stat(file_path)

                    return {
                        "success": True,
                        "file_path": file_path,
                        "content": content,
                        "size_bytes": stat.st_size,
                        "lines": len(content.splitlines()),
                        "partial_read": False
                    }
        except Exception as e:
            raise

    def _tool_write_file(self, file_path: str, content: str) -> Dict[str, Any]:
        """Write content to file"""
        try:
            # Resolve path relative to current directory
            if not os.path.isabs(file_path):
                file_path = os.path.join(self.current_directory, file_path)

            # Create directory if needed
            os.makedirs(os.path.dirname(file_path), exist_ok=True)

            with open(file_path, 'w', encoding='utf-8') as f:
                f.write(content)

            return {
                "success": True,
                "file_path": file_path,
                "bytes_written": len(content.encode('utf-8')),
                "lines_written": len(content.splitlines())
            }
        except Exception as e:
            raise

    def _tool_code_interpreter(self, code: str) -> Dict[str, Any]:
        """Execute Python code in restricted environment"""
        try:
            # Capture output
            import io
            import contextlib

            output_buffer = io.StringIO()
            error_buffer = io.StringIO()

            with contextlib.redirect_stdout(output_buffer), contextlib.redirect_stderr(error_buffer):
                exec(code)

            # Get output
            stdout = output_buffer.getvalue()
            stderr = error_buffer.getvalue()

            return {
                "success": True,
                "stdout": stdout,
                "stderr": stderr,
            }
        except Exception as e:
            raise

    def _tool_execute_command(self, command: str, working_dir: Optional[str] = None) -> Dict[str, Any]:
        """Execute shell command"""
        try:
            # Use current directory if not specified
            if working_dir is None:
                working_dir = self.current_directory
            elif not os.path.isabs(working_dir):
                working_dir = os.path.join(self.current_directory, working_dir)

            # Update current directory if 'cd' command
            if command.strip().startswith('cd '):
                new_dir = command.strip()[3:].strip()
                if not os.path.isabs(new_dir):
                    new_dir = os.path.join(self.current_directory, new_dir)

                if os.path.isdir(new_dir):
                    self.current_directory = os.path.abspath(new_dir)
                    return {
                        "success": True,
                        "command": command,
                        "output": f"Changed directory to: {self.current_directory}",
                        "return_code": 0
                    }
                else:
                    raise FileNotFoundError(f"Directory not found: {new_dir}")

            # Execute command
            result = subprocess.run(
                command,
                shell=True,
                capture_output=True,
                text=True,
                cwd=working_dir,
                timeout=30
            )

            return {
                "success": result.returncode == 0,
                "command": command,
                "output": result.stdout,
                "error": result.stderr if result.stderr else None,
                "return_code": result.returncode,
                "working_dir": working_dir
            }
        except subprocess.TimeoutExpired:
            raise TimeoutError(f"Command timed out after 30 seconds: {command}")
        except Exception as e:
            raise

    def _tool_rewrite_todo_list(self, items: List[str]) -> Dict[str, Any]:
        """Rewrite TODO list with new pending items"""
        # Keep completed and cancelled items
        kept_items = [
            item for item in self.todo_list
            if item.status in [TodoStatus.COMPLETED, TodoStatus.CANCELLED]
        ]

        # Create new pending items
        new_items = []
        for content in items:
            new_items.append(TodoItem(
                id=self.next_todo_id,
                content=content,
                status=TodoStatus.PENDING
            ))
            self.next_todo_id += 1

        # Update TODO list
        self.todo_list = kept_items + new_items

        return {
            "success": True,
            "kept_items": len(kept_items),
            "new_items": len(new_items),
            "total_items": len(self.todo_list)
        }

    def _tool_update_todo_status(self, updates: List[Dict[str, Any]]) -> Dict[str, Any]:
        """Update status of TODO items"""
        updated_count = 0

        for update in updates:
            item_id = update["id"]
            new_status = TodoStatus(update["status"])

            for item in self.todo_list:
                if item.id == item_id:
                    item.status = new_status
                    item.updated_at = datetime.now().isoformat()
                    updated_count += 1
                    break

        return {
            "success": True,
            "updated_items": updated_count,
            "total_items": len(self.todo_list)
        }

    def execute_task(self, task: str, max_iterations: int = 20) -> Dict[str, Any]:
        """
        Execute a task using available tools with system hints

        Args:
            task: The task to execute
            max_iterations: Maximum number of tool calls

        Returns:
            Task execution result
        """
        # Add timestamp to user message if enabled
        if self.config.enable_timestamps:
            timestamp_prefix = f"[{self._get_timestamp()}] "
            task = timestamp_prefix + task

        # Add user message
        self.conversation_history.append({"role": "user", "content": task})

        iteration = 0
        final_answer = None

        while iteration < max_iterations:
            iteration += 1
            logger.info(f"Iteration {iteration}/{max_iterations}")

            # Simulate time passing for demo
            self._advance_simulated_time(seconds=5)

            # Save trajectory at the start of each iteration
            self._save_trajectory(iteration)

            try:
                # Prepare messages for the model - add system hint as last user message
                messages_to_send = self.conversation_history.copy()
                system_hint = self._get_system_hint()
                if system_hint:
                    messages_to_send.append({"role": "user", "content": system_hint})

                # Store the messages being sent to LLM for trajectory logging
                self.last_llm_messages = messages_to_send

                # Call the model
                response = self.client.chat.completions.create(
                    model=self.model,
                    messages=messages_to_send,
                    tools=self._get_tools_description(),
                    tool_choice="auto",
                    temperature=_reasoning_safe_temperature(self.model, 0.3),
                    max_tokens=8192
                )

                message = response.choices[0].message
                has_tool_calls = bool(getattr(message, "tool_calls", None))

                # Terminal path: a text reply with no tool calls ends the loop,
                # even without the FINAL ANSWER: marker (e.g. a plain "hi"
                # reply). Previously only "FINAL ANSWER:" broke the loop, so
                # plain replies were re-sent for up to max_iterations.
                if not has_tool_calls:
                    self.conversation_history.append(message.model_dump())
                    content = (message.content or "").strip()
                    if content:
                        final_answer = (content.split("FINAL ANSWER:", 1)[1].strip()
                                        if "FINAL ANSWER:" in content else content)
                        logger.info(f"Terminal text response (no tool calls); final answer: {final_answer[:100]}...")
                    else:
                        logger.warning("Empty model response with no tool calls; "
                                       "stopping to avoid burning remaining iterations")
                    # Save final trajectory
                    self._save_trajectory(iteration, final_answer)
                    break

                # Handle tool calls
                if has_tool_calls:
                    self.conversation_history.append(message.model_dump())

                    for tool_call in message.tool_calls:
                        function_name = tool_call.function.name
                        function_args = json.loads(tool_call.function.arguments)

                        # Track tool call count
                        if self.config.enable_tool_counter:
                            self.tool_call_counts[function_name] = self.tool_call_counts.get(function_name, 0) + 1
                            call_number = self.tool_call_counts[function_name]
                        else:
                            call_number = 1

                        logger.info(f"Executing tool: {function_name} (call #{call_number})")

                        # Print tool arguments in a concise format
                        args_str = json.dumps(function_args)
                        if len(args_str) > 200:
                            logger.info(f"  📥 Args: {args_str[:200]}...")
                        else:
                            logger.info(f"  📥 Args: {args_str}")

                        # Execute the tool
                        result, error = self._execute_tool(function_name, function_args)

                        # Print tool result in a concise format
                        if error:
                            error_preview = str(error).replace('\n', ' ')[:150]
                            logger.info(f"  ❌ Error: {error_preview}")
                        else:
                            if isinstance(result, dict):
                                if result.get('success'):
                                    # Show key information for successful operations
                                    if 'output' in result and result['output']:
                                        output_preview = str(result['output']).replace('\n', ' ')[:100]
                                        logger.info(f"  ✅ Success: {output_preview}...")
                                    elif 'content' in result:
                                        # Handle read_file results
                                        if result.get('partial_read'):
                                            logger.info(f"  ✅ Success: Read lines {result.get('begin_line', 1)}-{result.get('end_line', 0)} "
                                                      f"({result.get('lines_read', 0)} lines) from {result.get('total_lines', 0)} total")
                                        else:
                                            logger.info(f"  ✅ Success: Read {result.get('lines', 0)} lines, {result.get('size_bytes', 0)} bytes")
                                    elif 'file_path' in result:
                                        logger.info(f"  ✅ Success: File operation on {result['file_path']}")
                                    else:
                                        logger.info(f"  ✅ Success: Operation completed")
                                elif result.get('success') is False:
                                    # Handle explicit failures (like binary file detection)
                                    if result.get('is_binary'):
                                        logger.info(f"  ⚠️ Binary file detected: {result.get('file_path', 'unknown')}")
                                    else:
                                        logger.info(f"  ⚠️ Failed: {result.get('error', 'Unknown error')[:100]}")
                                else:
                                    logger.info(f"  ✅ Success: Operation completed")
                            else:
                                result_preview = str(result).replace('\n', ' ')[:150]
                                logger.info(f"  ✅ Result: {result_preview}")

                        # Record tool call
                        tool_call_record = ToolCall(
                            tool_name=function_name,
                            arguments=function_args,
                            result=result if not error else None,
                            error=error,
                            call_number=call_number
                        )
                        self.tool_calls.append(tool_call_record)

                        # Prepare tool result message
                        tool_content = json.dumps(result)

                        # Add metadata to tool result if enabled
                        metadata_parts = []

                        if self.config.enable_timestamps:
                            metadata_parts.append(f"[{self._get_timestamp()}]")

                        if self.config.enable_tool_counter:
                            metadata_parts.append(f"[Tool call #{call_number} for '{function_name}']")

                        if metadata_parts:
                            tool_content = " ".join(metadata_parts) + "\n" + tool_content

                        # Add tool result
                        self.conversation_history.append({
                            "role": "tool",
                            "tool_call_id": tool_call.id,
                            "content": tool_content
                        })

                    # If the same turn also tagged FINAL ANSWER: (unusual with
                    # tool calls), still stop after recording the tool results.
                    if message.content and "FINAL ANSWER:" in message.content:
                        final_answer = message.content.split("FINAL ANSWER:", 1)[1].strip()
                        logger.info(f"Final answer found alongside tool calls: {final_answer[:100]}...")
                        self._save_trajectory(iteration, final_answer)
                        break

            except Exception as e:
                logger.error(f"Error during task execution: {str(e)}")
                # Save trajectory even on error
                self._save_trajectory(iteration)
                return {
                    "error": str(e),
                    "tool_calls": self.tool_calls,
                    "iterations": iteration,
                    "trajectory_file": self.config.trajectory_file if self.config.save_trajectory else None
                }

        # Save final trajectory before returning
        self._save_trajectory(iteration, final_answer)

        return {
            "final_answer": final_answer,
            "tool_calls": self.tool_calls,
            "todo_list": [
                {
                    "id": item.id,
                    "content": item.content,
                    "status": item.status.value
                }
                for item in self.todo_list
            ],
            "iterations": iteration,
            "success": final_answer is not None,
            "trajectory_file": self.config.trajectory_file if self.config.save_trajectory else None
        }

    def reset(self):
        """Reset the agent's state"""
        self.tool_call_counts = {}
        self.tool_calls = []
        self.todo_list = []
        self.next_todo_id = 1
        self.current_directory = os.getcwd()
        self.simulated_time = datetime.now()
        self.last_llm_messages = None
        self._init_system_prompt()
        logger.info("Agent state reset")

config.py

"""
Configuration module for System-Hint Enhanced Agent
"""

import os
from typing import Optional
from dataclasses import dataclass


@dataclass
class AgentConfig:
    """Configuration for the System-Hint Agent"""

    # API Configuration
    api_key: Optional[str] = None
    provider: str = "kimi"
    model: Optional[str] = None

    # System Hint Features
    enable_timestamps: bool = True
    enable_tool_counter: bool = True
    enable_todo_list: bool = True
    enable_detailed_errors: bool = True
    enable_system_state: bool = True

    # Formatting Options
    timestamp_format: str = "%Y-%m-%d %H:%M:%S"
    simulate_time_delay: bool = False

    # Execution Options
    max_iterations: int = 20
    verbose: bool = False
    timeout: int = 30  # seconds for command execution

    @classmethod
    def from_env(cls) -> "AgentConfig":
        """Create configuration from environment variables"""
        return cls(
            api_key=os.getenv("KIMI_API_KEY") or os.getenv("MOONSHOT_API_KEY") or os.getenv("OPENROUTER_API_KEY"),
            provider=os.getenv("LLM_PROVIDER", "kimi"),
            model=os.getenv("LLM_MODEL"),
            enable_timestamps=os.getenv("ENABLE_TIMESTAMPS", "true").lower() == "true",
            enable_tool_counter=os.getenv("ENABLE_TOOL_COUNTER", "true").lower() == "true",
            enable_todo_list=os.getenv("ENABLE_TODO_LIST", "true").lower() == "true",
            enable_detailed_errors=os.getenv("ENABLE_DETAILED_ERRORS", "true").lower() == "true",
            enable_system_state=os.getenv("ENABLE_SYSTEM_STATE", "true").lower() == "true",
            timestamp_format=os.getenv("TIMESTAMP_FORMAT", "%Y-%m-%d %H:%M:%S"),
            simulate_time_delay=os.getenv("SIMULATE_TIME_DELAY", "false").lower() == "true",
            max_iterations=int(os.getenv("MAX_ITERATIONS", "20")),
            verbose=os.getenv("VERBOSE", "false").lower() == "true",
            timeout=int(os.getenv("COMMAND_TIMEOUT", "30"))
        )

    def validate(self) -> bool:
        """Validate the configuration"""
        if not self.api_key:
            raise ValueError("API key is required. Set KIMI_API_KEY (or MOONSHOT_API_KEY / OPENROUTER_API_KEY fallback).")

        if self.provider not in ["kimi", "moonshot"]:
            raise ValueError(f"Unsupported provider: {self.provider}")

        if self.max_iterations < 1:
            raise ValueError("max_iterations must be at least 1")

        if self.timeout < 1:
            raise ValueError("timeout must be at least 1 second")

        return True


# Default configuration presets
PRESETS = {
    "full": AgentConfig(
        enable_timestamps=True,
        enable_tool_counter=True,
        enable_todo_list=True,
        enable_detailed_errors=True,
        enable_system_state=True
    ),
    "minimal": AgentConfig(
        enable_timestamps=False,
        enable_tool_counter=False,
        enable_todo_list=False,
        enable_detailed_errors=False,
        enable_system_state=False
    ),
    "debug": AgentConfig(
        enable_timestamps=True,
        enable_tool_counter=True,
        enable_todo_list=True,
        enable_detailed_errors=True,
        enable_system_state=True,
        verbose=True
    ),
    "demo": AgentConfig(
        enable_timestamps=True,
        enable_tool_counter=True,
        enable_todo_list=True,
        enable_detailed_errors=True,
        enable_system_state=True,
        simulate_time_delay=True
    )
}


def get_config(preset: Optional[str] = None) -> AgentConfig:
    """
    Get configuration from environment or preset

    Args:
        preset: Optional preset name ('full', 'minimal', 'debug', 'demo')

    Returns:
        AgentConfig instance
    """
    if preset and preset in PRESETS:
        config = PRESETS[preset]
        # Override with environment API key if available
        config.api_key = os.getenv("KIMI_API_KEY") or os.getenv("MOONSHOT_API_KEY") or os.getenv("OPENROUTER_API_KEY")
        return config

    return AgentConfig.from_env()

main.py

"""
Main entry point for System-Hint Enhanced Agent
Supports command-line tasks and interactive mode
"""

import os
import sys
import json
import logging
import argparse
from datetime import datetime
from pathlib import Path
from agent import SystemHintAgent, SystemHintConfig, TodoStatus

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


def print_section(title: str):
    """Print a formatted section header"""
    print("\n" + "="*80)
    print(f"  {title}")
    print("="*80)


def print_result(result: dict):
    """Print formatted result"""
    if result.get('success'):
        print("\n✅ Task completed successfully!")
        if result.get('final_answer'):
            print("\n📝 Final Answer:")
            print("-"*40)
            print(result['final_answer'])
    else:
        print("\n❌ Task failed!")
        if result.get('error'):
            print(f"Error: {result['error']}")

    print(f"\n📊 Statistics:")
    print(f"  - Iterations: {result.get('iterations', 0)}")
    print(f"  - Tool calls: {len(result.get('tool_calls', []))}")

    if result.get('trajectory_file'):
        print(f"\n💾 Trajectory saved to: {result['trajectory_file']}")

    if result.get('todo_list'):
        print(f"\n📋 Final TODO List:")
        for item in result['todo_list']:
            status_emoji = {
                'pending': '⏳',
                'in_progress': '🔄',
                'completed': '✅',
                'cancelled': '❌'
            }.get(item['status'], '❓')
            print(f"  [{item['id']}] {status_emoji} {item['content']} ({item['status']})")

    # Show tool call summary
    if result.get('tool_calls'):
        print(f"\n🔧 Tool Call Summary:")
        tool_summary = {}
        for call in result['tool_calls']:
            tool_name = call.tool_name
            if tool_name not in tool_summary:
                tool_summary[tool_name] = {
                    'count': 0,
                    'success': 0,
                    'failed': 0
                }
            tool_summary[tool_name]['count'] += 1
            if call.error:
                tool_summary[tool_name]['failed'] += 1
            else:
                tool_summary[tool_name]['success'] += 1

        for tool_name, stats in tool_summary.items():
            print(f"  - {tool_name}: {stats['count']} calls "
                  f"({stats['success']} success, {stats['failed']} failed)")


def get_sample_task() -> str:
    """Get the sample task for summarizing week1 and week2 projects"""
    return """Analyze and summarize the AI Agent projects in week1 and week2 directories. Create a comprehensive analysis file 'project_analysis_report.md' containing:

   - Overview of all the projects in week1 and week2 directories
   - What you have learned from the projects
    """


def execute_single_task(task: str, config: SystemHintConfig = None, verbose: bool = False,
                        provider: str = "kimi", model: str = None):
    """Execute a single task with the agent"""
    api_key = os.getenv("KIMI_API_KEY") or os.getenv("MOONSHOT_API_KEY") or os.getenv("OPENROUTER_API_KEY")
    if not api_key:
        print("❌ Error: Please set KIMI_API_KEY environment variable")
        print("   export KIMI_API_KEY='your-api-key-here'")
        print("   （如果只想离线查看状态栏效果，请运行 python main.py --mode preview）")
        return None

    if config is None:
        config = SystemHintConfig(
            enable_timestamps=True,
            enable_tool_counter=True,
            enable_todo_list=True,
            enable_detailed_errors=True,
            enable_system_state=True
        )

    agent = SystemHintAgent(
        api_key=api_key,
        provider=provider,
        model=model,
        config=config,
        verbose=verbose
    )

    # For project analysis tasks, navigate to parent directory
    if "week1" in task.lower() and "week2" in task.lower():
        agent.current_directory = str(Path(__file__).parent.parent)
        print(f"📁 Working directory set to: {agent.current_directory}")

    print("\n🚀 Executing task...")
    result = agent.execute_task(task, max_iterations=30)
    return result


def interactive_mode():
    """Run the agent in interactive mode"""
    print_section("Interactive Mode - System-Hint Agent")

    api_key = os.getenv("KIMI_API_KEY") or os.getenv("MOONSHOT_API_KEY") or os.getenv("OPENROUTER_API_KEY")
    if not api_key:
        print("❌ Error: Please set KIMI_API_KEY environment variable")
        print("   export KIMI_API_KEY='your-api-key-here'")
        return

    # Initialize agent with full features
    config = SystemHintConfig(
        enable_timestamps=True,
        enable_tool_counter=True,
        enable_todo_list=True,
        enable_detailed_errors=True,
        enable_system_state=True
    )

    agent = SystemHintAgent(
        api_key=api_key,
        provider="kimi",
        config=config,
        verbose=False
    )

    print("\n✅ Agent initialized with full system hints")
    print("\nAvailable commands:")
    print("  'sample' - Run the sample project analysis task")
    print("  'reset'  - Reset agent state and conversation")
    print("  'config' - Show current configuration")
    print("  'quit'   - Exit interactive mode")
    print("\nOr enter any task for the agent to complete.")

    while True:
        try:
            print("\n" + "-"*60)
            user_input = input("Task > ").strip()

            if not user_input:
                continue

            if user_input.lower() == 'quit':
                print("👋 Goodbye!")
                break

            elif user_input.lower() == 'sample':
                task = get_sample_task()
                print("\n📋 Running sample task:")
                print(task)

                # Navigate to parent directory for project analysis
                original_dir = agent.current_directory
                agent.current_directory = str(Path(__file__).parent.parent)

                result = agent.execute_task(task, max_iterations=100)
                print_result(result)

                # Restore directory
                agent.current_directory = original_dir

            elif user_input.lower() == 'reset':
                agent.reset()
                print("✅ Agent state reset")

            elif user_input.lower() == 'config':
                print("\n📋 Current Configuration:")
                print(f"  - Timestamps: {'✅' if config.enable_timestamps else '❌'}")
                print(f"  - Tool Counter: {'✅' if config.enable_tool_counter else '❌'}")
                print(f"  - TODO List: {'✅' if config.enable_todo_list else '❌'}")
                print(f"  - Detailed Errors: {'✅' if config.enable_detailed_errors else '❌'}")
                print(f"  - System State: {'✅' if config.enable_system_state else '❌'}")
                print(f"  - Current Directory: {agent.current_directory}")

            else:
                # Execute user task
                result = agent.execute_task(user_input, max_iterations=25)
                print_result(result)

        except KeyboardInterrupt:
            print("\n\n⚠️ Interrupted. Type 'quit' to exit.")
        except Exception as e:
            print(f"\n❌ Error: {str(e)}")
            logger.error(f"Error in interactive mode: {e}", exc_info=True)


def demo_basic_features():
    """Demonstrate basic system hint features"""
    print_section("Demo: Basic System Hint Features")

    api_key = os.getenv("KIMI_API_KEY") or os.getenv("MOONSHOT_API_KEY") or os.getenv("OPENROUTER_API_KEY")
    if not api_key:
        print("❌ Please set KIMI_API_KEY environment variable")
        return

    config = SystemHintConfig(
        enable_timestamps=True,
        enable_tool_counter=True,
        enable_todo_list=True,
        enable_detailed_errors=True,
        enable_system_state=True
    )

    agent = SystemHintAgent(
        api_key=api_key,
        provider="kimi",
        config=config,
        verbose=False
    )

    task = """Please complete the following tasks:
    1. Create a test directory called 'demo_output'
    2. Write a Python script that counts files in the current directory
    3. Execute the script and save the output
    4. Create a summary report of what was done

    Use the TODO list to track your progress."""

    result = agent.execute_task(task)
    print_result(result)


def demo_tool_loop_prevention():
    """Demonstrate tool call loop prevention"""
    print_section("Demo: Tool Call Loop Prevention")

    api_key = os.getenv("KIMI_API_KEY") or os.getenv("MOONSHOT_API_KEY") or os.getenv("OPENROUTER_API_KEY")
    if not api_key:
        print("❌ Please set KIMI_API_KEY environment variable")
        return

    config = SystemHintConfig(
        enable_timestamps=False,
        enable_tool_counter=True,
        enable_todo_list=False,
        enable_detailed_errors=True,
        enable_system_state=False
    )

    agent = SystemHintAgent(
        api_key=api_key,
        provider="kimi",
        config=config,
        verbose=False
    )

    task = """Try to read a file called 'nonexistent_file.txt' up to 3 times.
    After each failed attempt, note the failure and stop after 3 attempts."""

    result = agent.execute_task(task, max_iterations=10)
    print_result(result)

    if result.get('tool_calls'):
        read_file_calls = [c for c in result['tool_calls'] if c.tool_name == 'read_file']
        print(f"\n🛡️ Tool counter prevented loop: {len(read_file_calls)} read_file attempts")
        for call in read_file_calls:
            print(f"  - Call #{call.call_number}: {'Failed' if call.error else 'Success'}")


def demo_comparison():
    """Compare with and without system hints"""
    print_section("Demo: System Hints Comparison")

    api_key = os.getenv("KIMI_API_KEY") or os.getenv("MOONSHOT_API_KEY") or os.getenv("OPENROUTER_API_KEY")
    if not api_key:
        print("❌ Please set KIMI_API_KEY environment variable")
        return

    task = """Create a simple Python script that prints 'Hello World' and save it as 'hello.py'."""

    # With system hints
    print("\n📋 WITH System Hints:")
    config_with = SystemHintConfig(
        enable_timestamps=True,
        enable_tool_counter=True,
        enable_todo_list=True,
        enable_detailed_errors=True,
        enable_system_state=True
    )

    agent_with = SystemHintAgent(
        api_key=api_key,
        provider="kimi",
        config=config_with,
        verbose=False
    )

    result_with = agent_with.execute_task(task, max_iterations=10)
    print(f"  - Success: {result_with.get('success')}")
    print(f"  - Iterations: {result_with.get('iterations')}")
    print(f"  - Tool calls: {len(result_with.get('tool_calls', []))}")

    # Without system hints
    print("\n📋 WITHOUT System Hints:")
    config_without = SystemHintConfig(
        enable_timestamps=False,
        enable_tool_counter=False,
        enable_todo_list=False,
        enable_detailed_errors=False,
        enable_system_state=False
    )

    agent_without = SystemHintAgent(
        api_key=api_key,
        provider="kimi",
        config=config_without,
        verbose=False
    )

    result_without = agent_without.execute_task(task, max_iterations=10)
    print(f"  - Success: {result_without.get('success')}")
    print(f"  - Iterations: {result_without.get('iterations')}")
    print(f"  - Tool calls: {len(result_without.get('tool_calls', []))}")

    print("\n💡 System hints typically lead to more efficient task completion!")


def preview_status_bar(config: SystemHintConfig):
    """离线预览：展示五种状态栏（system hint）技术如何改变模型看到的上下文。

    对应书中实验 2-8 的五种技术。整个过程在本地渲染，**不发起任何 LLM 调用，
    因此无需 API Key**。每个案例都做一次“无状态栏 vs 有状态栏”的对照，
    直观地展示 Agent 框架在上下文末尾注入的显式状态。
    """
    from datetime import datetime as _dt

    print_section("离线预览：Agent 状态栏（System Hint）如何改变上下文")
    print(
        "说明：以下每个案例对比【无状态栏】（模型只能看到原始轨迹）与\n"
        "      【有状态栏】（框架把隐式状态提炼成显式知识注入上下文末尾）。\n"
        "      场景取自书中的 Xfinity 退款案例，全部在本地渲染，不调用任何 API。"
    )

    # 用占位 Key 构造 Agent；构造过程不联网。固定模拟时间以便输出稳定可复现。
    agent = SystemHintAgent(
        api_key="offline-preview",
        provider="kimi",
        config=config,
        verbose=False,
    )
    agent.config.simulate_time_delay = True
    agent.simulated_time = _dt(2025, 9, 14, 10, 30, 45)

    enabled = []

    # --- 案例 1：时间戳跟踪 -------------------------------------------------
    if config.enable_timestamps:
        enabled.append("时间戳跟踪")
        print("\n【案例 1 · 时间戳跟踪】为用户消息与工具结果加上时间前缀")
        follow_up = "Can you call them again to follow up?"
        print("-" * 60)
        print("  无状态栏：" + follow_up)
        print("  有状态栏：" + f"[{agent._get_timestamp()}] " + follow_up)
        print("  → Agent 能理解“昨天的文件”与“今天的修改”之间的时序关系。")

    # --- 案例 2：工具调用计数器 -------------------------------------------
    if config.enable_tool_counter:
        enabled.append("工具调用计数器")
        print("\n【案例 2 · 工具调用计数器】在工具结果上标注第几次调用")
        raw_result = json.dumps({"success": True, "output": "Call connected, no answer"})
        # 复用 execute_task 中的元信息拼装格式
        metadata = []
        if config.enable_timestamps:
            metadata.append(f"[{agent._get_timestamp()}]")
        metadata.append("[Tool call #3 for 'phone_call']")
        print("-" * 60)
        print("  无状态栏：" + raw_result)
        print("  有状态栏：" + " ".join(metadata) + "\n            " + raw_result)
        print("  → 显式计数触发模型的模式识别：到达 3/3 上限时主动停止，不再重复拨打。")

    # --- 案例 3：TODO 列表管理 -------------------------------------------
    if config.enable_todo_list:
        enabled.append("TODO 列表管理")
        print("\n【案例 3 · TODO 列表管理】把多步任务分解并持续复述")
        agent._tool_rewrite_todo_list(items=[
            "拨打 Xfinity 客服核实退款政策",
            "提交退款申请",
            "确认退款到账",
        ])
        agent._tool_update_todo_status(updates=[
            {"id": 1, "status": "completed"},
            {"id": 2, "status": "in_progress"},
        ])
        print("-" * 60)
        print("  无状态栏：（模型需自行从长轨迹中回忆还剩哪些子任务，易遗漏）")
        print("  有状态栏：")
        for line in agent._format_todo_list().splitlines():
            print("    " + line)
        print("  → TODO 列表充当外部记忆，确保行动与总体规划保持一致。")

    # --- 案例 4：详细错误信息 -------------------------------------------
    if config.enable_detailed_errors:
        enabled.append("详细错误信息")
        print("\n【案例 4 · 详细错误信息】把裸异常升级为带修复建议的诊断")
        exc = FileNotFoundError("File not found: /home/user/refund_policy.txt")
        detailed = agent._get_detailed_error(
            exc, "read_file", {"file_path": "refund_policy.txt"}
        )
        print("-" * 60)
        print("  无状态栏：" + str(exc))
        print("  有状态栏：")
        for line in detailed.splitlines():
            print("    " + line)
        print("  → Agent 从盲目重试转向分析性的问题解决（验证路径、检查目录、用绝对路径）。")

    # --- 案例 5：系统状态感知 -------------------------------------------
    if config.enable_system_state:
        enabled.append("系统状态感知")
        print("\n【案例 5 · 系统状态感知】注入当前时间、目录、操作系统、Shell、Python 版本")
        print("-" * 60)
        print("  无状态栏：（模型不知道自己身处哪个目录、哪种操作系统）")
        print("  有状态栏：")
        for line in agent._get_system_state().splitlines():
            print("    " + line)
        print("  → 操作系统信息让 Agent 做出平台相关决策（Linux 用 apt、macOS 用 brew）。")

    # --- 汇总：实际注入上下文末尾的完整状态栏 ---------------------------
    print_section("实际追加到上下文末尾的状态栏（一条 role=user 的消息）")
    hint = agent._get_system_hint()
    if hint:
        print(hint)
        print(
            "\n注意：这条消息的 role 是 user，但内容由 Agent 框架自动生成，"
            "追加在上下文最末尾——\n紧邻模型即将生成的新 token，因此获得最高注意力权重；"
            "且因为是“追加”而非\n“修改”，前面已缓存的 KV Cache 前缀不受影响。"
        )
    else:
        print("（当前配置下系统状态与 TODO 均被禁用，无状态栏可注入。）")

    print("\n本次预览启用的技术：" + ("、".join(enabled) if enabled else "（全部禁用）"))
    print("提示：用 --no-timestamps / --no-counter / --no-todo / --no-errors / --no-state")
    print("      可分别关闭某一类，观察上下文的差异。")


def main():
    """Main function with command-line argument support"""
    parser = argparse.ArgumentParser(
        description=(
            "System-Hint Enhanced AI Agent（对应书中实验 2-8 “Agent 状态栏”）\n"
            "演示五种状态栏技术如何把上下文里的隐式状态提炼为显式知识，"
            "从而改变 Agent 的行为。"
        ),
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=(
            "示例：\n"
            "  # 离线预览状态栏效果（无需 API Key，推荐先跑这个）\n"
            "  python main.py --mode preview\n"
            "  # 只看某一类技术的前后对比（例如关闭其它四类）\n"
            "  python main.py --mode preview --no-todo --no-errors --no-state --no-timestamps\n"
            "  # 用真实模型执行单个任务（需要 KIMI_API_KEY）\n"
            "  python main.py --mode single --task \"创建一个 hello world 脚本\"\n"
            "  # 对比“启用/禁用状态栏”的实际执行效果\n"
            "  python main.py --mode demo --demo comparison\n"
        ),
    )

    parser.add_argument(
        "--mode",
        choices=["preview", "single", "interactive", "demo", "sample"],
        default="interactive",
        help=(
            "执行模式：preview=离线预览状态栏（无需 API Key）；"
            "single=执行单个任务；interactive=交互模式（默认）；"
            "demo=运行内置演示；sample=运行示例任务"
        )
    )

    parser.add_argument(
        "--task",
        type=str,
        help="要执行的任务描述（single 模式必填）"
    )

    parser.add_argument(
        "--demo",
        choices=["basic", "loop", "comparison"],
        help="指定要运行的演示（demo 模式）：basic=综合演示，loop=循环防护，comparison=启用/禁用状态栏对比"
    )

    parser.add_argument(
        "--provider",
        type=str,
        default="kimi",
        help="LLM 提供方（默认：kimi，兼容 moonshot）"
    )

    parser.add_argument(
        "--model",
        type=str,
        default=None,
        help="模型名称覆盖（默认由 provider 决定，如 kimi-k3）"
    )

    parser.add_argument(
        "--output",
        type=str,
        default=None,
        help="轨迹输出文件路径（默认：trajectory.json）"
    )

    parser.add_argument(
        "--no-timestamps",
        action="store_true",
        help="关闭时间戳跟踪"
    )

    parser.add_argument(
        "--no-counter",
        action="store_true",
        help="关闭工具调用计数器"
    )

    parser.add_argument(
        "--no-todo",
        action="store_true",
        help="关闭 TODO 列表管理"
    )

    parser.add_argument(
        "--no-errors",
        action="store_true",
        help="关闭详细错误信息"
    )

    parser.add_argument(
        "--no-state",
        action="store_true",
        help="关闭系统状态感知"
    )

    parser.add_argument(
        "--verbose",
        action="store_true",
        help="输出详细日志"
    )

    args = parser.parse_args()

    # Configure based on command-line flags
    config = SystemHintConfig(
        enable_timestamps=not args.no_timestamps,
        enable_tool_counter=not args.no_counter,
        enable_todo_list=not args.no_todo,
        enable_detailed_errors=not args.no_errors,
        enable_system_state=not args.no_state
    )
    if args.output:
        config.trajectory_file = args.output

    print("\n" + "🤖"*40)
    print("  SYSTEM-HINT ENHANCED AGENT")
    print("🤖"*40)

    if args.mode == "preview":
        preview_status_bar(config)

    elif args.mode == "single":
        if not args.task:
            print("❌ Error: --task required for single mode")
            print("Example: python main.py --mode single --task 'Create a hello world script'")
            sys.exit(1)

        result = execute_single_task(args.task, config, verbose=args.verbose,
                                     provider=args.provider, model=args.model)
        if result:
            print_result(result)

    elif args.mode == "sample":
        # Run the sample task
        task = get_sample_task()
        print("\n📋 Running sample task:")
        print("-"*60)
        print(task)
        print("-"*60)

        result = execute_single_task(task, config, verbose=args.verbose,
                                     provider=args.provider, model=args.model)
        if result:
            print_result(result)

    elif args.mode == "demo":
        if args.demo == "basic":
            demo_basic_features()
        elif args.demo == "loop":
            demo_tool_loop_prevention()
        elif args.demo == "comparison":
            demo_comparison()
        else:
            # Run all demos
            print("\nRunning all demonstrations...")
            demo_basic_features()
            input("\nPress Enter to continue...")
            demo_tool_loop_prevention()
            input("\nPress Enter to continue...")
            demo_comparison()

    else:  # interactive mode
        interactive_mode()

    print("\n👋 Thank you for using System-Hint Enhanced Agent!")


if __name__ == "__main__":
    main()

openrouter_fallback.py

"""Universal OpenRouter fallback helper (Chapter 2 experiments).

Goal: every experiment keeps working when the direct provider key is missing
but ``OPENROUTER_API_KEY`` is present. Default behavior is fully preserved:

  * If a primary provider key is present -> use the primary provider unchanged.
  * Else if ``OPENROUTER_API_KEY`` is present -> route through OpenRouter
    (base_url=https://openrouter.ai/api/v1) and translate the model id.
  * Else raise a clear error listing every accepted key.

Model translation (only applied when the OpenRouter fallback is active):
  * ids already containing "/" are passed through unchanged;
  * gpt-* / o1* / o3* / o4* / chatgpt* -> "openai/<id>";
  * claude-*                            -> "anthropic/claude-opus-4.8";
  * kimi-*                              -> "moonshotai/kimi-k2.6";
  * anything else                       -> passed through unchanged.
"""

import os

OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"


def is_openrouter_key(api_key):
    """OpenRouter keys reliably start with ``sk-or-``."""
    return bool(api_key) and api_key.startswith("sk-or-")


def map_model_to_openrouter(model):
    """Translate a bare provider model id into an OpenRouter-qualified id."""
    if not model or "/" in model:
        return model
    low = model.lower()
    if low.startswith(("gpt-", "gpt5", "o1", "o3", "o4", "chatgpt")):
        return "openai/" + model
    if low.startswith("claude"):
        return "anthropic/claude-opus-4.8"
    if low.startswith("kimi"):
        return "moonshotai/kimi-k2.6"
    return model


def resolve_llm(model=None, primary_keys=("OPENAI_API_KEY",), primary_base_url=None):
    """Resolve ``(api_key, base_url, model)`` honoring the primary->OpenRouter fallback.

    Args:
        model: requested model id (may be remapped when the fallback activates).
        primary_keys: env var names checked in order for the primary provider.
        primary_base_url: base_url used when a primary key is found
            (``None`` means the OpenAI SDK default / official endpoint).
    """
    or_key = os.getenv("OPENROUTER_API_KEY")
    # gpt-5.x (incl. gpt-5.6*) needs OpenAI org-verification on the direct API;
    # when an OpenRouter key is present, prefer routing these ids through it.
    if or_key and model and model.lower().startswith("gpt-5"):
        return or_key, OPENROUTER_BASE_URL, map_model_to_openrouter(model)
    for env in primary_keys:
        key = os.getenv(env)
        if key:
            return key, primary_base_url, model
    if or_key:
        return or_key, OPENROUTER_BASE_URL, map_model_to_openrouter(model)
    accepted = ", ".join(list(primary_keys) + ["OPENROUTER_API_KEY"])
    raise RuntimeError(
        "No LLM API key found. Set one of the primary keys or the universal "
        "fallback. Accepted keys: " + accepted + ". See env.example."
    )

quickstart.py

"""
Quick Start for System-Hint Enhanced Agent
Simple demonstration using the sample task
"""

import os
import sys
from pathlib import Path
from agent import SystemHintAgent, SystemHintConfig


def main():
    """Quick start demonstration"""

    # Check for API key
    api_key = os.getenv("KIMI_API_KEY") or os.getenv("MOONSHOT_API_KEY") or os.getenv("OPENROUTER_API_KEY")
    if not api_key:
        print("❌ Error: Please set KIMI_API_KEY environment variable")
        print("\nSetup instructions:")
        print("  1. Copy env.example to .env")
        print("  2. Add your Kimi API key to .env")
        print("  3. Run: export KIMI_API_KEY='your-api-key-here'")
        return

    print("\n" + "="*60)
    print("  🚀 System-Hint Agent Quick Start")
    print("="*60)

    # Sample task for analyzing projects
    task = """Analyze and summarize the AI Agent projects in week1 and week2 directories:

1. Navigate to the parent directory to access both week1 and week2 folders
2. For week1 directory:
   - List all project folders
   - Read the README.md or main.py from the 'context' project
   - Identify the key concepts implemented
3. For week2 directory:
   - List all project folders  
   - Read the README.md from one project
   - Understand the advanced features
4. Create a brief summary file 'quick_summary.txt' with your findings

Use the TODO list to organize and track your analysis steps."""

    print("\n📋 Task:")
    print("-"*60)
    print(task)
    print("-"*60)

    # Create agent with all features enabled
    print("\n🔧 Initializing agent with full system hints...")
    config = SystemHintConfig(
        enable_timestamps=True,
        enable_tool_counter=True,
        enable_todo_list=True,
        enable_detailed_errors=True,
        enable_system_state=True
    )

    agent = SystemHintAgent(
        api_key=api_key,
        provider="kimi",
        config=config,
        verbose=False  # Set to True to see full API interactions
    )

    # Set working directory to parent to access week1/week2
    agent.current_directory = str(Path(__file__).parent.parent)
    print(f"📁 Working directory: {agent.current_directory}")

    print("\n🚀 Executing task (this may take a moment)...")
    print("-"*60)

    # Execute the task
    result = agent.execute_task(task, max_iterations=25)

    # Display results
    print("\n" + "="*60)
    print("  📊 Results")
    print("="*60)

    if result.get('success'):
        print("\n✅ Task completed successfully!")

        if result.get('final_answer'):
            print("\n📝 Summary:")
            print("-"*40)
            answer = result['final_answer']
            # Display first 800 chars for readability
            if len(answer) > 800:
                print(answer[:800] + "...\n[Output truncated for display]")
            else:
                print(answer)
    else:
        print("\n⚠️ Task did not complete fully")
        if result.get('error'):
            print(f"Error: {result['error']}")

    # Statistics
    print("\n📈 Execution Statistics:")
    print(f"  • Iterations: {result.get('iterations', 0)}")
    print(f"  • Tool calls: {len(result.get('tool_calls', []))}")

    # Tool usage breakdown
    if result.get('tool_calls'):
        tool_counts = {}
        for call in result['tool_calls']:
            tool_counts[call.tool_name] = tool_counts.get(call.tool_name, 0) + 1

        print("\n🔧 Tools Used:")
        for tool, count in sorted(tool_counts.items(), key=lambda x: x[1], reverse=True):
            print(f"  • {tool}: {count} call{'s' if count > 1 else ''}")

    # TODO list summary
    if result.get('todo_list'):
        completed = sum(1 for item in result['todo_list'] if item['status'] == 'completed')
        total = len(result['todo_list'])
        print(f"\n📋 TODO Progress: {completed}/{total} tasks completed")

        # Show first few TODO items
        print("\nTODO Items:")
        for item in result['todo_list'][:5]:
            status_symbol = {
                'pending': '⏳',
                'in_progress': '🔄',
                'completed': '✅',
                'cancelled': '❌'
            }.get(item['status'], '❓')
            # Truncate long content
            content = item['content']
            if len(content) > 60:
                content = content[:57] + "..."
            print(f"  {status_symbol} [{item['id']}] {content}")

        if total > 5:
            print(f"  ... and {total - 5} more items")

    print("\n" + "="*60)
    print("  ✨ Quick Start Complete!")
    print("="*60)
    print("\n💡 Tips:")
    print("  • Run 'python main.py' for interactive mode")
    print("  • Run 'python main.py --mode sample' to run this task again")
    print("  • Run 'python main.py --mode demo' for more demonstrations")
    print("  • Set verbose=True in the agent for detailed API logs")
    print("\n")


if __name__ == "__main__":
    main()

test_basic.py

"""
Basic test to verify System-Hint Agent functionality
"""

import os
import sys
from agent import SystemHintAgent, SystemHintConfig, TodoStatus

def test_basic_functionality():
    """Test basic agent functionality without API calls"""
    print("Testing System-Hint Agent components...")

    # Test configuration
    config = SystemHintConfig(
        enable_timestamps=True,
        enable_tool_counter=True,
        enable_todo_list=True,
        enable_detailed_errors=True,
        enable_system_state=True
    )
    print("✅ Configuration created successfully")

    # Test agent initialization (without API key for basic test)
    try:
        agent = SystemHintAgent(
            api_key="test-key",  # Dummy key for initialization test
            provider="kimi",
            config=config,
            verbose=False
        )
        print("✅ Agent initialized successfully")
    except Exception as e:
        print(f"❌ Agent initialization failed: {e}")
        return False

    # Test tool implementations (without API calls)
    print("\nTesting tool implementations:")

    # Test file operations
    test_file = "test_output.txt"
    try:
        # Test write_file
        result = agent._tool_write_file(test_file, "Test content")
        assert result["success"] == True
        print("✅ write_file tool works")

        # Test read_file
        result = agent._tool_read_file(test_file)
        assert result["success"] == True
        assert "Test content" in result["content"]
        print("✅ read_file tool works")

        # Clean up
        os.remove(test_file)

    except Exception as e:
        print(f"❌ File operation test failed: {e}")

    # Test code interpreter
    try:
        result = agent._tool_code_interpreter("result = 2 + 2")
        assert result["success"] == True
        assert result["result"] == 4
        print("✅ code_interpreter tool works")
    except Exception as e:
        print(f"❌ Code interpreter test failed: {e}")

    # Test TODO list operations
    try:
        # Test rewrite_todo_list
        result = agent._tool_rewrite_todo_list(["Task 1", "Task 2", "Task 3"])
        assert result["success"] == True
        assert result["new_items"] == 3
        print("✅ rewrite_todo_list tool works")

        # Test update_todo_status
        result = agent._tool_update_todo_status([
            {"id": 1, "status": "completed"},
            {"id": 2, "status": "in_progress"}
        ])
        assert result["success"] == True
        assert result["updated_items"] == 2
        print("✅ update_todo_status tool works")

        # Verify TODO list state
        assert len(agent.todo_list) == 3
        assert agent.todo_list[0].status == TodoStatus.COMPLETED
        assert agent.todo_list[1].status == TodoStatus.IN_PROGRESS
        print("✅ TODO list management works correctly")

    except Exception as e:
        print(f"❌ TODO list test failed: {e}")

    # Test system state
    try:
        state = agent._get_system_state()
        assert "Current Time:" in state
        assert "Current Directory:" in state
        assert "System:" in state
        print("✅ System state tracking works")
    except Exception as e:
        print(f"❌ System state test failed: {e}")

    # Test error handling
    try:
        # This should fail and generate detailed error
        result = agent._tool_read_file("/nonexistent/file.txt")
    except Exception as e:
        error_detail = agent._get_detailed_error(e, "read_file", {"file_path": "/nonexistent/file.txt"})
        assert "FileNotFoundError" in str(e.__class__.__name__) or "No such file" in str(e)
        assert "Suggestions:" in error_detail
        print("✅ Detailed error handling works")

    print("\n✅ All basic tests passed!")
    return True

def test_command_execution():
    """Test command execution tool"""
    print("\nTesting command execution:")

    config = SystemHintConfig(enable_detailed_errors=True)
    agent = SystemHintAgent(
        api_key="test-key",
        provider="kimi",
        config=config,
        verbose=False
    )

    try:
        # Test simple command
        result = agent._tool_execute_command("echo 'Hello, World!'")
        assert result["success"] == True
        assert "Hello, World!" in result["output"]
        print("✅ Command execution works")

        # Test directory change
        original_dir = agent.current_directory
        result = agent._tool_execute_command("cd /tmp")
        assert agent.current_directory == "/tmp"
        agent.current_directory = original_dir  # Restore
        print("✅ Directory tracking works")

    except Exception as e:
        print(f"⚠️ Command execution test skipped: {e}")

    return True

if __name__ == "__main__":
    print("="*60)
    print("  System-Hint Agent Component Tests")
    print("="*60)

    # Run basic tests
    if test_basic_functionality():
        test_command_execution()
        print("\n✨ All tests completed successfully!")
    else:
        print("\n❌ Some tests failed")
        sys.exit(1)

test_hint_behavior.py

#!/usr/bin/env python
"""
Test script to demonstrate that system hints are added as user messages
temporarily before sending to LLM, but not stored in conversation history.
"""

import os
import json
from agent import SystemHintAgent, SystemHintConfig

def test_hint_behavior():
    """Test and demonstrate the system hint behavior"""
    api_key = os.getenv("KIMI_API_KEY")
    if not api_key:
        print("❌ Please set KIMI_API_KEY environment variable")
        return

    # Create agent with system hints enabled
    config = SystemHintConfig(
        enable_timestamps=True,
        enable_system_state=True,
        enable_todo_list=True,
        save_trajectory=True,
        trajectory_file="test_hint_trajectory.json"
    )

    agent = SystemHintAgent(
        api_key=api_key,
        provider="kimi",
        config=config,
        verbose=False
    )

    # Execute a simple task
    task = "Create a file called test.txt with content 'Testing hint behavior'"
    result = agent.execute_task(task, max_iterations=5)

    if result['success']:
        print("✅ Task completed successfully\n")

    # Analyze the conversation history
    print("=" * 60)
    print("CONVERSATION HISTORY ANALYSIS")
    print("=" * 60)

    # Load the saved trajectory
    with open("test_hint_trajectory.json", 'r') as f:
        trajectory = json.load(f)

    conversation = trajectory['conversation_history']

    print(f"\nTotal messages in conversation history: {len(conversation)}")
    print("\nMessage roles and previews:")

    for i, msg in enumerate(conversation, 1):
        role = msg['role']
        content = msg.get('content', '')

        # Check if content contains system hints
        has_system_state = 'SYSTEM STATE' in content
        has_todo_list = 'CURRENT TASKS' in content

        preview = content[:80].replace('\n', ' ')
        if len(content) > 80:
            preview += "..."

        print(f"\n{i}. [{role.upper()}]")
        print(f"   Preview: {preview}")

        if has_system_state or has_todo_list:
            print(f"   ⚠️ Contains system hints: System State={has_system_state}, TODO List={has_todo_list}")

    # Summary
    print("\n" + "=" * 60)
    print("SUMMARY")
    print("=" * 60)

    # Check if any messages contain system hints
    hints_in_history = any(
        'SYSTEM STATE' in msg.get('content', '') or 
        'CURRENT TASKS' in msg.get('content', '')
        for msg in conversation
    )

    if hints_in_history:
        print("❌ System hints found in conversation history (unexpected!)")
    else:
        print("✅ No system hints stored in conversation history (expected behavior)")
        print("   System hints are added as temporary user messages before LLM calls")
        print("   but are NOT stored in the conversation history.")

    # Clean up test files
    if os.path.exists("test.txt"):
        os.remove("test.txt")

    print("\n" + "=" * 60)

if __name__ == "__main__":
    test_hint_behavior()

test_read_file.py

test_trajectory_logging.py

verify_trajectory.py

view_trajectory.py

#!/usr/bin/env python
"""
Utility to view saved trajectory files
Usage: python view_trajectory.py [trajectory_file]
"""

import json
import sys
from pathlib import Path
from datetime import datetime

def format_time(iso_string):
    """Convert ISO timestamp to readable format"""
    try:
        dt = datetime.fromisoformat(iso_string.replace('Z', '+00:00'))
        return dt.strftime("%Y-%m-%d %H:%M:%S")
    except:
        return iso_string

def view_trajectory(file_path="trajectory.json"):
    """View trajectory file with formatted output"""
    try:
        with open(file_path, 'r') as f:
            data = json.load(f)

        print("\n" + "="*80)
        print("TRAJECTORY ANALYSIS")
        print("="*80)

        print(f"\n📅 Timestamp: {format_time(data['timestamp'])}")
        print(f"🤖 Model: {data['model']}")
        print(f"🔄 Total Iterations: {data['iteration']}")
        print(f"💬 Conversation Messages: {len(data['conversation_history'])}")
        print(f"🔧 Tool Calls: {len(data['tool_calls'])}")

        if data['final_answer']:
            print(f"\n✅ Task Completed Successfully")
            print(f"Final Answer Preview: {data['final_answer'][:200]}...")
        else:
            print(f"\n⚠️ Task Not Completed")

        # Show tool calls breakdown
        if data['tool_calls']:
            print("\n📊 Tool Usage Summary:")
            tool_counts = {}
            for call in data['tool_calls']:
                name = call['tool_name']
                tool_counts[name] = tool_counts.get(name, 0) + 1

            for tool, count in tool_counts.items():
                print(f"  - {tool}: {count} call(s)")

        # Show TODO list if present
        if data['todo_list']:
            print("\n📋 TODO List:")
            for item in data['todo_list']:
                status_symbol = {
                    'pending': '⏳',
                    'in_progress': '🔄', 
                    'completed': '✅',
                    'cancelled': '❌'
                }.get(item['status'], '❓')
                print(f"  [{item['id']}] {status_symbol} {item['content']}")

        # Show conversation snippets
        print("\n💬 Conversation Highlights:")
        for i, msg in enumerate(data['conversation_history'][:5], 1):
            role = msg['role']
            content = msg.get('content', '')
            if content:
                preview = content[:100].replace('\n', ' ')
                if len(content) > 100:
                    preview += "..."
                print(f"  {i}. [{role}] {preview}")

        if len(data['conversation_history']) > 5:
            print(f"  ... and {len(data['conversation_history']) - 5} more messages")

        print("\n" + "="*80)
        print(f"Full trajectory saved in: {file_path}")
        print(f"File size: {Path(file_path).stat().st_size:,} bytes")
        print("="*80 + "\n")

    except FileNotFoundError:
        print(f"❌ Error: Trajectory file '{file_path}' not found")
        print("Run an agent task first to generate a trajectory file.")
    except json.JSONDecodeError:
        print(f"❌ Error: Invalid JSON in '{file_path}'")
    except Exception as e:
        print(f"❌ Error reading trajectory: {e}")

if __name__ == "__main__":
    file_path = sys.argv[1] if len(sys.argv) > 1 else "trajectory.json"
    view_trajectory(file_path)

run_sample.sh

#!/bin/bash

# Sample script to demonstrate System-Hint Agent usage
# This script runs the sample task that analyzes week1 and week2 projects

echo "================================"
echo "System-Hint Agent Sample Task"
echo "================================"
echo ""
echo "This will analyze and summarize the AI Agent projects"
echo "in week1 and week2 directories using the System-Hint Agent."
echo ""

# Check if KIMI_API_KEY is set
if [ -z "$KIMI_API_KEY" ]; then
    echo "❌ Error: KIMI_API_KEY environment variable is not set"
    echo ""
    echo "Please set it using:"
    echo "  export KIMI_API_KEY='your-api-key-here'"
    echo ""
    exit 1
fi

echo "✅ KIMI_API_KEY is configured"
echo ""
echo "Starting analysis..."
echo "----------------------------------------"

# Run the sample task
python main.py --mode sample

echo ""
echo "Sample task completed!"
echo ""
echo "You can also try:"
echo "  - Interactive mode: python main.py"
echo "  - Custom task: python main.py --mode single --task 'Your task here'"
echo "  - Demos: python main.py --mode demo"
echo ""

trajectory.json

{
  "timestamp": "2026-07-18T10:55:08.962158",
  "iteration": 1,
  "provider": "kimi",
  "model": "moonshotai/kimi-k2.6",
  "conversation_history": [
    {
      "role": "system",
      "content": "You are an intelligent assistant with access to various tools for file operations, code execution, and system commands.\n\nYour task is to complete the given objectives efficiently using the available tools. Think step by step and use tools as needed.\n\n## TODO List Management Rules:\n- For any complex task with 3+ distinct steps, immediately create a TODO list using `rewrite_todo_list`\n- Break down the user's request into specific, actionable TODO items\n- Update TODO items to 'in_progress' when starting work on them using `update_todo_status`\n- Mark items as 'completed' immediately after finishing them\n- Only have ONE item 'in_progress' at a time\n- If you encounter errors or need to change approach, update relevant TODOs to 'cancelled' and add new ones\n- Use the TODO list as your primary planning and tracking mechanism\n- Reference TODO items by their ID when discussing progress\n\n## Key Behaviors:\n1. ALWAYS start complex tasks by creating a TODO list\n2. Pay attention to timestamps to understand the timeline of events\n3. Notice tool call numbers (e.g., \"Tool call #3\") to avoid repetitive loops - if you see high numbers, change strategy\n4. Learn from detailed error messages to fix issues and adapt your approach\n5. Be aware of your current directory and system environment shown in system state\n6. When exploring projects, systematically read key files (README, main.py, agent.py) to understand structure\n\n## Error Handling:\n- Read error messages carefully - they contain specific information about what went wrong\n- Use the suggestions provided in error messages to fix issues\n- If a tool fails multiple times (check the call number), try a different approach\n- Common fixes: check file paths, verify current directory, ensure proper permissions\n\nImportant: When you have completed all tasks, clearly state \"FINAL ANSWER:\" followed by a comprehensive summary of what was accomplished."
    },
    {
      "role": "user",
      "content": "[2026-07-18 10:55:04] 计算 3+5，不要使用任何工具，直接在回复正文里输出一行：FINAL ANSWER: <结果>"
    },
    {
      "content": "FINAL ANSWER: 8",
      "refusal": null,
      "role": "assistant",
      "annotations": null,
      "audio": null,
      "function_call": null,
      "tool_calls": null,
      "reasoning": "用户要求计算 3+5，并且明确说不要使用任何工具，直接在回复正文里输出一行：FINAL ANSWER: <结果>。\n\n这是一个非常简单的数学问题，3+5=8。用户特别要求不使用工具，所以我应该直接回答。\n\n格式要求：FINAL ANSWER: 8",
      "reasoning_details": [
        {
          "type": "reasoning.text",
          "text": "用户要求计算 3+5，并且明确说不要使用任何工具，直接在回复正文里输出一行：FINAL ANSWER: <结果>。\n\n这是一个非常简单的数学问题，3+5=8。用户特别要求不使用工具，所以我应该直接回答。\n\n格式要求：FINAL ANSWER: 8",
          "format": "unknown",
          "index": 0
        }
      ]
    }
  ],
  "last_llm_messages": [
    {
      "role": "system",
      "content": "You are an intelligent assistant with access to various tools for file operations, code execution, and system commands.\n\nYour task is to complete the given objectives efficiently using the available tools. Think step by step and use tools as needed.\n\n## TODO List Management Rules:\n- For any complex task with 3+ distinct steps, immediately create a TODO list using `rewrite_todo_list`\n- Break down the user's request into specific, actionable TODO items\n- Update TODO items to 'in_progress' when starting work on them using `update_todo_status`\n- Mark items as 'completed' immediately after finishing them\n- Only have ONE item 'in_progress' at a time\n- If you encounter errors or need to change approach, update relevant TODOs to 'cancelled' and add new ones\n- Use the TODO list as your primary planning and tracking mechanism\n- Reference TODO items by their ID when discussing progress\n\n## Key Behaviors:\n1. ALWAYS start complex tasks by creating a TODO list\n2. Pay attention to timestamps to understand the timeline of events\n3. Notice tool call numbers (e.g., \"Tool call #3\") to avoid repetitive loops - if you see high numbers, change strategy\n4. Learn from detailed error messages to fix issues and adapt your approach\n5. Be aware of your current directory and system environment shown in system state\n6. When exploring projects, systematically read key files (README, main.py, agent.py) to understand structure\n\n## Error Handling:\n- Read error messages carefully - they contain specific information about what went wrong\n- Use the suggestions provided in error messages to fix issues\n- If a tool fails multiple times (check the call number), try a different approach\n- Common fixes: check file paths, verify current directory, ensure proper permissions\n\nImportant: When you have completed all tasks, clearly state \"FINAL ANSWER:\" followed by a comprehensive summary of what was accomplished."
    },
    {
      "role": "user",
      "content": "[2026-07-18 10:55:04] 计算 3+5，不要使用任何工具，直接在回复正文里输出一行：FINAL ANSWER: <结果>"
    },
    {
      "role": "user",
      "content": "=== SYSTEM STATE ===\nCurrent Time: 2026-07-18 10:55:04\nCurrent Directory: /Users/boj/book/ai-agent-book/chapter2/system-hint\nSystem: Darwin (25.3.0)\nShell Environment: macOS Terminal (zsh/bash)\nPython Version: 3.11.4\n"
    }
  ],
  "tool_calls": [],
  "todo_list": [],
  "current_directory": "/Users/boj/book/ai-agent-book/chapter2/system-hint",
  "final_answer": "8",
  "config": {
    "enable_timestamps": true,
    "enable_tool_counter": true,
    "enable_todo_list": true,
    "enable_detailed_errors": true,
    "enable_system_state": true,
    "timestamp_format": "%Y-%m-%d %H:%M:%S",
    "simulate_time_delay": false
  }
}