agent-task-executor/README.md

Agent Task Executor

A flexible task execution framework that uses LLMs (Large Language Models).

Features

• Unique task identification
• State tracking and monitoring
• Checkpoint creation and rollback
• Resource usage monitoring
• Error handling and retry mechanism
• Execution path tracking
• Detailed status reporting

Project Structure

agent_task_executor/
├── config/                     # Configuration files and utilities
│   ├── config_loader.py       # Configuration loading utilities
│   ├── llm_config.yaml        # LLM configuration settings
│   ├── secure_config.py       # Secure configuration handling
│   └── secure.json           # Secure configuration data (encrypted)
│
├── tasksamples/               # Sample task implementations
│   ├── __init__.py
│   └── text_analysis_task.py  # Text analysis task implementation
│
├── tests/                     # Test suite
│   ├── test_config.py        # Configuration tests
│   └── test_llm.py           # LLM functionality tests
│
├── scripts/                   # Utility scripts
│
├── llm_client.py             # LLM API client implementation
├── llm_executor.py           # Core LLM execution logic
├── task_executor.py          # Task execution framework
├── sample_task.py            # Sample task demonstration
│
├── requirements.txt          # Python dependencies
├── pyproject.toml           # Project metadata and build configuration
└── .env.example             # Example environment variables

Core Components

1. Task Execution Framework

   • task_executor.py: Core framework for task execution
   • llm_executor.py: LLM-specific execution logic
   • llm_client.py: Client for interacting with LLM APIs

2. Configuration Management

   • config/: Contains all configuration-related files
   • Secure configuration handling with encryption support
   • LLM-specific configurations

3. Task Implementations

   • tasksamples/: Directory for task implementations
   • Text analysis task with Chinese text support
   • Extensible for additional task types

4. Testing

   • Comprehensive test suite for configuration and LLM functionality
   • Unit tests for core components

Setup and Dependencies

The project uses Python with the following key dependencies:

• httpx: For making HTTP requests
• tenacity: For retry logic
• pyyaml: For configuration handling
• cryptography: For secure configuration

Environment Management

Package Management with UV

The project uses uv for fast Python package management:

# Install uv (if not already installed)
pip install uv

# Create virtual environment and install dependencies
uv venv
uv pip install -r requirements.txt

# Activate virtual environment
# On Windows:
.venv\Scripts\activate
# On Unix or MacOS:
source .venv/bin/activate

Lock File

The project uses uv.lock to ensure dependency consistency:

• uv.lock: Contains exact versions of all dependencies
• To update dependencies: uv pip compile requirements.txt -o uv.lock
• To sync with lock file: uv pip sync uv.lock

Environment Variables

Required environment variables:

• DEEPSEEK_API_KEY: API key for DeepSeek LLM service
• DEEPSEEK_API_BASE: Base URL for DeepSeek API
• CONFIG_PATH: Path to configuration directory (default: ./config)

Create a .env file in the project root:

DEEPSEEK_API_KEY=your_api_key_here
DEEPSEEK_API_BASE=https://api.deepseek.com/v1
CONFIG_PATH=./config

Configuration Files

1. llm_config.yaml: LLM service configuration

   llm:
     provider: deepseek
     model: deepseek-chat
     temperature: 0.7
     max_tokens: 2000
   

2. secure.json: Encrypted configuration for sensitive data

   • Generated and managed by secure_config.py
   • Uses encryption for storing API keys and secrets

Usage

1. Install dependencies:

uv pip install -r requirements.txt

2. Run the sample task:

python sample_task.py

3. Create your own task executor:

from task_executor import TaskExecutor

class CustomTaskExecutor(TaskExecutor):
    def __init__(self):
        super().__init__()
        # Define your task steps
        
    def validate_input(self, input_data):
        # Implement input validation
        pass
        
    def execute_step(self, step_id, step_data):
        # Implement step execution
        pass

Task Execution Framework

Core Components

1. Task Status Tracking

• Unique task ID
• Step-by-step progress monitoring
• Resource usage tracking
• Execution path recording

2. Checkpointing System

• Automatic checkpoint creation every 5 steps
• Rollback capability
• State preservation

3. Error Handling

• Maximum 3 retry attempts
• Timeout monitoring (300 seconds)
• Detailed error logging

4. Status Reporting

• Real-time status updates
• Resource usage statistics
• Execution path history
• Error and warning logs

Status Output Format

{
    "task_id": "unique_id",
    "timestamp": "ISO-8601 timestamp",
    "current_step": {
        "id": "step_id",
        "name": "step_name",
        "status": "status",
        "progress": "percentage"
    },
    "checkpoints": ["checkpoint_list"],
    "resources": {
        "used": "used_resources",
        "available": "available_resources"
    },
    "execution_path": ["executed_steps"],
    "next_actions": ["possible_actions"]
}

密钥管理

1. 创建了安全的配置管理系统： SecureConfig 类用于加密存储敏感信息 使用 Fernet 对称加密 适当的文件权限管理（Windows 和 Unix）
2. 实现了配置加载器： 从 YAML 文件加载基本配置 从加密存储加载敏感信息 支持环境变量覆盖
3. 建立了清晰的配置优先级： API 密钥：环境变量 > 安全存储 > YAML 配置 API 基础 URL：环境变量 > API 配置 > 提供商默认值
4. 完成了全面的测试覆盖： 安全配置的加密/解密 文件权限测试 配置优先级测试 API 设置测试
5. 添加了必要的依赖： cryptography 用于加密 pywin32 用于 Windows 安全性 PyYAML 用于配置文件
6. 这个实现提供了： 更好的安全性：敏感信息使用加密存储 灵活性：多种配置方式和清晰的优先级 跨平台支持：同时支持 Windows 和 Unix 良好的测试覆盖

测试

pytest tests/test_config.py -v

Contributing

Development Setup

1. Clone the repository:

git clone http://gitea.towards-agi.cn/zhukang/agent-task-executor.git
cd agent-task-executor

2. Set up the development environment:

# Install uv if not installed
pip install uv

# Create virtual environment and install dependencies
uv venv
uv pip install -e ".[dev]"

3. Run tests:

python -m pytest

Code Style

This project uses:

• black for code formatting
• isort for import sorting
• mypy for type checking

Format your code before committing:

# Format code
black .
# Sort imports
isort .
# Type check
mypy .

Creating New Tasks

1. Create a new task class in tasksamples/ that inherits from TaskExecutor
2. Define task steps in the constructor
3. Implement step handlers
4. Add tests in tests/

Example:

from agent_task_executor.task_executor import TaskExecutor, TaskStatus

class MyTaskExecutor(TaskExecutor):
    def __init__(self):
        super().__init__(llm_model="deepseek-chat")
        self.task_steps = [
            {
                "id": "step1",
                "name": "Step One",
                "required_info": ["input_data"],
                "instruction": "Process the input data."
            },
            # Add more steps...
        ]

    async def handle_step1(self, step_input: dict) -> dict:
        # Implement step logic
        return {"result": "processed data"}

Pull Request Process

1. Create a new branch for your feature
2. Write tests for new functionality
3. Update documentation as needed
4. Submit a pull request with a clear description of changes

Commit Messages

Follow the Conventional Commits specification:

• feat: New features
• fix: Bug fixes
• docs: Documentation changes
• test: Adding or updating tests
• refactor: Code refactoring
• chore: Maintenance tasks

Example:

feat: add new text classification task

License

MIT License