คู่มือ OpenClaw Subagent

{
  "version": "1.0.0",
  "name": "OpenClaw Project",
  "description": "การตั้งค่าระบบ OpenClaw Subagent",
  
  // ตั้งค่า coordinator
  "coordinator": {
    "task_interval_ms": 1000,
    "max_tasks": 100,
    "timeout_ms": 30000,
    "retry_count": 3,
    "log_level": "info"
  },

  // ตั้งค่า subagents
  "agents": {
    "researcher": {
      "enabled": true,
      "max_parallel": 3,
      "tools": ["web_search", "api_call", "data_scraper"],
      "api_keys": ["OPENAI_API_KEY", "SERPAPI_KEY"]
    },
    
    "writer": {
      "enabled": true,
      "max_parallel": 2,
      "tools": ["text_generation", "markdown_conversion"],
      "default_model": "gpt-4"
    },
    
    "analyst": {
      "enabled": true,
      "max_parallel": 2,
      "tools": ["data_analysis", "pattern_recognition"],
      "max_data_points": 10000
    },
    
    "developer": {
      "enabled": true,
      "max_parallel": 2,
      "tools": ["code_generation", "debugging", "testing"],
      "supported_languages": ["python", "javascript", "typescript"]
    }
  },

  // ตั้งค่า output
  "output": {
    "directory": "./output",
    "format": "markdown",
    "include_timestamp": true,
    "compact_output": false
  },

  // ตั้งค่า logging
  "logging": {
    "enabled": true,
    "level": "INFO",
    "directory": "./logs",
    "max_file_size_mb": 50
  },

  // ตั้งค่า security
  "security": {
    "rate_limit_ms": 100,
    "max_concurrent_requests": 10,
    "allow_http": false
  }
}

{
  "version": "1.0.0",
  "name": "OpenClaw Production Config",
  "description": "Configuration สำหรับ production environment",

  "coordinator": {
    "task_interval_ms": 500,
    "max_tasks": 500,
    "timeout_ms": 60000,
    "retry_count": 5,
    "log_level": "info"
  },

  "agents": {
    "researcher": {
      "enabled": true,
      "max_parallel": 5,
      "tools": ["web_search", "api_call", "data_scraper", "database_query"],
      "api_keys": ["OPENAI_API_KEY", "SERPAPI_KEY", "GOOGLE_CSE_ID"],
      "rate_limit": {
        "requests_per_minute": 60
      }
    },

    "writer": {
      "enabled": true,
      "max_parallel": 3,
      "tools": ["text_generation", "markdown_conversion", "code_documentation"],
      "default_model": "gpt-4-turbo",
      "temperature": 0.7,
      "max_tokens": 4096
    },

    "analyst": {
      "enabled": true,
      "max_parallel": 4,
      "tools": ["data_analysis", "pattern_recognition", "statistical_analysis"],
      "max_data_points": 50000,
      "time_window_seconds": 3600
    },

    "developer": {
      "enabled": true,
      "max_parallel": 3,
      "tools": ["code_generation", "debugging", "testing", "code_review"],
      "supported_languages": ["python", "javascript", "typescript", "java", "go"],
      "code_generation": {
        "max_lines": 500,
        "include_tests": true,
        "inline_comments": true
      }
    }
  },

  "output": {
    "directory": "/var/www/outputs/openclaw",
    "format": "markdown",
    "include_timestamp": true,
    "compact_output": false,
    "separate_logs": true
  },

  "logging": {
    "enabled": true,
    "level": "INFO",
    "directory": "/var/log/openclaw",
    "max_file_size_mb": 100,
    "rotate_count": 5,
    "json_format": true
  },

  "security": {
    "rate_limit_ms": 200,
    "max_concurrent_requests": 20,
    "allow_http": false,
    "allowed_origins": ["https://example.com"],
    "api_secret_key": "${SECRET_KEY_ENV}"
  },

  "monitoring": {
    "enabled": true,
    "metrics_endpoint": "/metrics",
    "health_check_interval_ms": 30000,
    "alert_thresholds": {
      "task_timeout_rate_percent": 10,
      "error_rate_percent": 5
    }
  }
}

{
  "workflow": {
    "sequential": {
      "enabled": true,
      "stages": [
        {
          "name": "research",
          "agent": "researcher",
          "input_transform": "raw_query",
          "output_transform": "structured_data"
        },
        {
          "name": "analysis",
          "agent": "analyst",
          "input_transform": "raw_data",
          "output_transform": "insights"
        },
        {
          "name": "writing",
          "agent": "writer",
          "input_transform": "insights",
          "output_transform": "content"
        },
        {
          "name": "development",
          "agent": "developer",
          "input_transform": "content_spec",
          "output_transform": "executable_code"
        }
      ],
      "transition_delay_ms": 500,
      "on_error": "stop"  // or "continue"
    }
  }
}

import json
from openclaw import OpenClaw, Coordinator

# Load configuration
with open('openclaw.json') as f:
    config = json.load(f)

# Initialize OpenClaw
client = OpenClaw(config)

# Create a simple task
task = client.create_task(
    name='research_summary',
    description='Research latest AI trends and summarize',
    input_data={
        'topic': 'AI Subagents',
        'keywords': ['OpenClaw', 'Architecture']
    }
)

# Execute task
result = client.execute(task)
print(result.summary)
print(result.aggregated_data)

import asyncio
from openclaw import OpenClaw, Task

# Initialize client
client = OpenClaw('openclaw.json')

# Create multiple research tasks
topics = ['AI Agents', 'Parallel Processing', 
          'Workflow Management', 'System Architecture']

async def run_parallel_research():
    tasks = []
    for topic in topics:
        task = Task(
            agent='researcher',
            action='gather_info',
            input_data={'topic': topic},
            priority=2
        )
        tasks.append(task)
    
    # Execute all tasks in parallel
    # This will spawn multiple researcher agents simultaneously
    results = await client.execute_parallel(tasks, max_parallel=4)
    
    # Analyze and combine results
    combined_results = {
        'topics': topics,
        'total_researched': len(results),
        'data_points': sum(r.data_points for r in results)
    }
    
    return combined_results

# Run the parallel execution
if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    final_results = loop.run_until_complete(run_parallel_research())
    print(json.dumps(final_results, indent=2))

from openclaw import OpenClaw, Workflow, Task

# Initialize client
client = OpenClaw('openclaw.json')

# Define sequential workflow
workflow = Workflow(name='content_pipeline')

# Step 1: Research
workflow.add_stage(
    agent='researcher',
    action='gather_research',
    name='research',
    input_transform={'topic': 'input.topic'}
)

# Step 2: Analysis
workflow.add_stage(
    agent='analyst',
    action='analyze_data',
    name='analysis',
    input_transform={'data': 'research.output'}
)

# Step 3: Writing
workflow.add_stage(
    agent='writer',
    action='write_content',
    name='writing',
    input_transform={'insights': 'analysis.output'}
)

# Step 4: Code Generation
workflow.add_stage(
    agent='developer',
    action='generate_code',
    name='code',
    input_transform={'spec': 'writing.output'}
)

# Execute the complete workflow
result = client.execute_workflow(
    workflow=workflow,
    input_data={'topic': 'AI Subagents'}
)

print("Research:", result.stages.research.output)
print("Content:", result.stages.writing.output)
print("Code:", result.stages.code.output)

from openclaw import OpenClaw, HierarchicalClient

# Initialize hierarchical client
# This creates a master coordinator with multiple sub-coordinators
client = HierarchicalClient(
    name='production-hierarchy',
    num_groups=3,
    agents_per_group=5
)

# The client automatically creates a tree structure:
# Master Coordinator (1)
#   Group Coordinator A (3)
#     Workers 1-5 (5 each)
#   Group Coordinator B (3)
#     Workers 6-10 (5 each)
#   Group Coordinator C (3)
#     Workers 11-15 (5 each)

# Submit tasks to the hierarchical system
tasks = []
for i in range(15):
    task = {
        'name': f'task_{i}',
        'type': 'analysis',
        'priority': i % 3,
        'data': {'id': i}
    }
    tasks.append(task)

# Execute with automatic hierarchical scheduling
results = client.execute_hierarchical(tasks)

# Check group distribution
print("Group A tasks:", len(results.groups.group_a))
print("Group B tasks:", len(results.groups.group_b))
print("Group C tasks:", len(results.groups.group_c))

# Access specialized expert agents when needed
if results.overall_status == 'error':
    expert_result = client.call_expert_agent(
        task_id=results.failed_tasks[0],
        expert_type='debugger'
    )
    print("Expert debug:", expert_result.fix)

from openclaw.agents import Researcher, Writer, Analyst, Developer

# === RESEARCHER EXAMPLE ===
researcher = Researcher(api_keys=['OPENAI_KEY', 'SERPAPI_KEY'])

research_result = researcher.run_research(
    topic='Latest AI Trends 2025',
    sources=['web', 'academic', 'news'],
    filters={
        'date_range': {'from': '2024-01-01'},
        'lang': ['en', 'th']
    }
)

# === WRITER EXAMPLE ===
writer = Writer(model='gpt-4', temperature=0.7)

content = writer.generate_content(
    content_type='blog_post',
    topic='OpenClaw Subagent',
    outline=research_result.outline,
    tone='professional',
    target_audience='developers',
    include_examples=True
)

# === ANALYST EXAMPLE ===
analyst = Analyst()

analysis_results = analyst.analyze_data(
    data=research_result.data,
    methods=['trends', 'clustering', 'correlation'],
    output_format='json'
)

# === DEVELOPER EXAMPLE ===
developer = Developer(supported_languages=['python', 'typescript'])

code = developer.generate_code(
    specification=content.specification,
    language='python',
    framework='fastapi',
    include_tests=True,
    inline_comments=True,
    max_lines=500
)

# Run tests
test_results = developer.run_tests(code='generated_code.py')

from openclaw import BaseAgent, AgentConfig
from typing import Dict, Any, Optional
import json

# Custom Agent Example: SEO Optimizer
class SEOOptimizer(BaseAgent):
    """Custom agent สำหรับวิเคราะห์และปรับปรุง SEO"""
    
    _agent_type = 'seo_optimizer'
    
    def __init__(self, config: AgentConfig):
        super().__init__(config)
        self.keywords_database = {}  # Load keywords from database
    
    async def analyze(self, content: str) -> Dict[str, Any]:
        # วิเคราะห์ SEO metrics
        metrics = {
            'keyword_density': await self._calculate_keyword_density(content),
            'readability_score': await self._calculate_readability(content),
            'semantic_structure': await self._analyze_semantic(content),
            'seo_score': await self._calculate_seo_score(content)
        }
        return metrics
    
    async def recommend(self, metrics: Dict[str, Any]) -> Dict[str, Any]:
        # สร้าง recommendations
        recommendations = []
        
        if metrics['keyword_density'] < 0.01:
            recommendations.append('Increase keyword density')
        
        if metrics['readability_score'] < 60:
            recommendations.append('Improve readability')
        
        return {
            'metrics': metrics,
            'recommendations': recommendations,
            'priority': await self._prioritize(recommendations)
        }
    
    async def optimize(self, content: str) -> str:
        # Auto-optimize content
        return await self._rewrite_with_seo(content)
    
    async def _calculate_keyword_density(self, content: str) -> float:
        # Implementation
        return 0.05  # Example density
    
    async def _calculate_readability(self, content: str) -> int:
        # Implementation
        return 75  # Example score
    
    async def _analyze_semantic(self, content: str) -> Dict[str, Any]:
        # Implementation
        return {'topics': ['AI', 'Technology']}
    
    async def _calculate_seo_score(self, content: str) -> int:
        # Calculate overall SEO score
        keywords_density = await self._calculate_keyword_density(content)
        readability = await self._calculate_readability(content)
        
        score = (keywords_density * 1000 + readability) / 2
        return int(score)

from openclaw import OpenClaw
import requests
import json

# Integration with PostgreSQL
class PostgreSQLWriter:
    """Writer agent ที่บันทึกผลลัพธ์ลง PostgreSQL"""
    
    def __init__(self, dsn: str):
        import asyncpg
        self.dsn = dsn
        self.conn = None
    
    async def connect(self):
        self.conn = await asyncpg.connect(self.dsn)
    
    async def save_result(self, result: Dict):
        await self.conn.execute(
            '''INSERT INTO results (data, created_at) VALUES ($1, NOW())''',
            json.dumps(result)
        )
    
    async def close(self):
        if self.conn:
            await self.conn.close()

# Integration with Slack Notifications
class SlackNotifier:
    """ส่งการแจ้งเตือนไปยัง Slack"""
    
    def __init__(self, webhook_url: str):
        self.webhook_url = webhook_url
    
    async def send_task_update(self, task_id: str, status: str, message: str):
        payload = {
            'text': f'{task_id}: {status}',
            'attachments': [{
                'color': '#36a64f' if status == 'success' else '#ff0000',
                'text': message
            }]
        }
        await requests.post(self.webhook_url, json=payload)

# Integration with Redis Queue
class RedisQueueClient:
    """Use Redis for task queuing"""
    
    def __init__'localhost', port: int = 6379):
        import redis
        self.redis = redis.Redis(host=host, port=port, decode_responses=True)
    
    async def enqueue_task(self, task: Dict):
        self.redis.lpush('openclaw:tasks', json.dumps(task))
    
    async def dequeue_task(self) -> Optional[Dict]:
        task = self.redis.rpop('openclaw:tasks')
        return json.loads(task) if task else None
    
    async mark_task_completed(self, task_id: str, result: Dict):
        self.redis.hset('openclaw:results', task_id, json.dumps(result))

import asyncio
import time
from openclaw import OpenClaw, PerformanceMetrics

# Performance monitoring decorator
def performance_monitor(func):
    async def wrapper(*args, **kwargs):
        start_time = time.time()
        try:
            result = await func(*args, **kwargs)
            execution_time = time.time() - start_time
            PerformanceMetrics.record_execution(func.__name__, execution_time)
            return result
        except Exception as e:
            PerformanceMetrics.record_error(func.__name__, str(e))
            raise
    return wrapper

# Agent pooling for better performance
class AgentPool:
    """Pool ของ agents เพื่อลด overhead ในการสร้าง agent ใหม่"""
    
    def __init__(self, agent_class, pool_size: int = 5):
        self.pool_size = pool_size
        self.available_agents = []
        self.in_use_agents = []
        self._initialize_pool(agent_class)
    
    def _initialize_pool(self, agent_class):
        for _ in range(self.pool_size):
            self.available_agents.append(agent_class())
    
    async def get_agent(self):
        if self.available_agents:
            agent = self.available_agents.pop()
            self.in_use_agents.append(agent)
            return agent
        return None  # Or wait and retry
    
    async return_agent(self, agent):
        if agent in self.in_use_agents:
            self.in_use_agents.remove(agent)
            self.available_agents.append(agent)

# Async batch processing
async def process_batch(tasks: list, max_concurrent: int = 10):
    semaphore = asyncio.Semaphore(max_concurrent)
    
    async def sem_task(task):
        async with semaphore:
            return await process_single_task(task)
    
    results = await asyncio.gather(*[sem_task(task) for task in tasks])
    return results