تسجيل التتبعات لخطوات الوكيل

لماذا تُعد سجلات التتبع ضرورية للوكلاء

إعداد وحدة التسجيل في Python

import logging
import sys

logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s [%(levelname)s] %(name)s: %(message)s',
    datefmt='%H:%M:%S',
    stream=sys.stdout
)

logger = logging.getLogger('myagent')

# Usage:
logger.debug('Step 1: reasoning started')
logger.info('Agent task completed in 5 steps')
logger.warning('Tool returned empty result')
logger.error('Failed to parse tool arguments')

# Output:
# 14:32:01 [DEBUG] myagent: Step 1: reasoning started
# 14:32:03 [INFO] myagent: Agent task completed in 5 steps

تسجيل كل خطوة من خطوات الاستدلال

import logging
import json

logger = logging.getLogger('myagent')

def log_step(step: int, thought: str, tool_name: str, tool_args: dict):
    logger.debug(
        f'Step {step}: '
        f'reasoning="{thought[:100]}" '
        f'tool={tool_name} '
        f'args={json.dumps(tool_args, ensure_ascii=False)[:200]}'
    )

# Example usage in the agent loop:
# log_step(
#     step=1,
#     thought='I need to find the current weather in Tokyo',
#     tool_name='get_weather',
#     tool_args={'city': 'Tokyo', 'unit': 'celsius'}
# )

if __name__ == '__main__':
    import sys
    logging.basicConfig(level=logging.DEBUG, format='%(message)s', stream=sys.stdout)
    log_step(
        step=1,
        thought='I need to find the current weather in Tokyo',
        tool_name='get_weather',
        tool_args={'city': 'Tokyo', 'unit': 'celsius'}
    )

تسجيل نتائج الأدوات

import logging

logger = logging.getLogger('myagent')

def log_tool_result(step: int, tool_name: str, result: str, success: bool):
    status = 'OK' if success else 'ERROR'
    preview = str(result)[:200].replace('\n', ' ')
    logger.debug(
        f'Step {step} result [{status}]: tool={tool_name} '
        f'result_preview="{preview}"'
    )

    if not success:
        logger.warning(f'Tool {tool_name} failed at step {step}')

# Log at the start of the step:
# log_step(step, thought, tool_name, tool_args)
# result = execute_tool(tool_name, tool_args)
# log_tool_result(step, tool_name, result, success=True)

if __name__ == '__main__':
    import sys
    logging.basicConfig(level=logging.DEBUG, format='%(message)s', stream=sys.stdout)
    log_tool_result(1, 'get_weather', '{"temp_c": 18, "condition": "cloudy"}', success=True)
    log_tool_result(2, 'get_weather', 'Connection timed out', success=False)

التسجيل المنظم بتنسيق JSON

import logging
import json
import sys

class JSONFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        log_obj = {
            'timestamp': self.formatTime(record),
            'level': record.levelname,
            'logger': record.name,
            'message': record.getMessage()
        }
        # Add any extra fields attached to the log record
        if hasattr(record, 'step'):
            log_obj['step'] = record.step
        if hasattr(record, 'tool'):
            log_obj['tool'] = record.tool
        return json.dumps(log_obj)

handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JSONFormatter())
logger = logging.getLogger('agent_trace')
logger.addHandler(handler)
logger.setLevel(logging.DEBUG)

logger.setLevel(logging.DEBUG)
logger.debug('Step 3: tool=search_web', extra={'step': 3, 'tool': 'search_web'})

التسجيل باستخدام الحقول الإضافية

import logging

logger = logging.getLogger('agent_trace')

def log_step_structured(step: int, tool: str, thought: str, args: dict):
    logger.debug(
        f'Step {step}: tool={tool}',
        extra={
            'step': step,
            'tool': tool,
            'thought': thought[:200],
            'tool_args': args
        }
    )

# If using a JSON formatter, this produces:
# {
#   'timestamp': '14:32:01',
#   'level': 'DEBUG',
#   'message': 'Step 3: tool=search_web',
#   'step': 3,
#   'tool': 'search_web',
#   'thought': 'I need to find recent news about...',
#   'args': {'query': 'AI news 2025'}
# }

if __name__ == '__main__':
    import sys
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(logging.Formatter('%(message)s | step=%(step)s tool=%(tool)s'))
    logger.addHandler(handler)
    logger.setLevel(logging.DEBUG)
    log_step_structured(3, 'search_web', 'I need to find recent news about...', {'query': 'AI news 2025'})

التسجيل في ملف

import logging
from logging.handlers import RotatingFileHandler
import sys

logger = logging.getLogger('myagent')
logger.setLevel(logging.DEBUG)

# Console handler — INFO and above
console = logging.StreamHandler(sys.stdout)
console.setLevel(logging.INFO)
console.setFormatter(logging.Formatter('%(message)s'))

# File handler — DEBUG and above, rotates at 10MB
file_handler = RotatingFileHandler(
    'agent_trace.log',
    maxBytes=10 * 1024 * 1024,  # 10 MB
    backupCount=3
)
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(logging.Formatter(
    '%(asctime)s [%(levelname)s] %(message)s'
))

logger.addHandler(console)
logger.addHandler(file_handler)

logger.info('Agent task completed in 5 steps')
logger.debug('Step 1: reasoning started')

تسجيل معرّفات الجلسات للوكلاء متعددي المستخدمين

import logging
import uuid

class SessionLogger:
    def __init__(self, name: str):
        self.logger = logging.getLogger(name)
        self.session_id = str(uuid.uuid4())[:8]

    def debug(self, msg: str, **kwargs):
        self.logger.debug(f'[session={self.session_id}] {msg}', **kwargs)

    def info(self, msg: str, **kwargs):
        self.logger.info(f'[session={self.session_id}] {msg}', **kwargs)

    def error(self, msg: str, **kwargs):
        self.logger.error(f'[session={self.session_id}] {msg}', **kwargs)

# Each agent run gets its own logger with a unique session ID
# log = SessionLogger('myagent')
# log.info(f'Starting task: {query}')  # [session=a3f1b290] Starting task: ...

if __name__ == '__main__':
    import sys
    logging.basicConfig(level=logging.INFO, format='%(message)s', stream=sys.stdout)
    log = SessionLogger('myagent')
    log.info(f'Starting task: summarize the quarterly report')

قياس زمن كل خطوة

import time
import logging

logger = logging.getLogger('myagent')

def timed_tool_call(tool_name: str, tool_fn, args: dict) -> str:
    start = time.perf_counter()
    try:
        result = tool_fn(**args)
        elapsed = time.perf_counter() - start
        logger.debug(f'Tool {tool_name} completed in {elapsed:.2f}s')
        return result
    except Exception as e:
        elapsed = time.perf_counter() - start
        logger.error(f'Tool {tool_name} failed in {elapsed:.2f}s: {e}')
        raise

# In the agent loop:
# result = timed_tool_call('search_web', search_web, {'query': 'Python'})
# Logs: Tool search_web completed in 1.34s

if __name__ == '__main__':
    import sys
    logging.basicConfig(level=logging.DEBUG, format='%(message)s', stream=sys.stdout)
    def search_web(query):
        return f'3 results for {query}'
    result = timed_tool_call('search_web', search_web, {'query': 'Python'})
    print('Tool result:', result)

النمط الكامل لتتبع الخطوات

import time
import logging
import json

logger = logging.getLogger('myagent')

def trace_step(step_num: int, thought: str, tool: str, args: dict, execute_fn):
    # Log decision
    logger.debug(
        f'Step {step_num}: thought="{thought[:80]}" tool={tool} '
        f'args={json.dumps(args)[:100]}'
    )

    # Execute with timing
    t0 = time.perf_counter()
    try:
        result = execute_fn(tool, args)
        elapsed = time.perf_counter() - t0
        preview = str(result)[:100].replace('\n', ' ')
        logger.debug(f'Step {step_num} done in {elapsed:.2f}s: "{preview}"')
        return result
    except Exception as e:
        elapsed = time.perf_counter() - t0
        logger.error(f'Step {step_num} failed in {elapsed:.2f}s: {e}')
        return f'ERROR: {e}'

if __name__ == '__main__':
    import sys
    logging.basicConfig(level=logging.DEBUG, format='%(message)s', stream=sys.stdout)
    def execute_fn(tool, args):
        return f'42 (from {tool})'
    trace_step(1, 'I should compute the answer', 'calculator', {'expr': '6*7'}, execute_fn)

إيقاف السجلات في بيئة الإنتاج

import os
import logging
import sys

# Read log level from environment variable
log_level_str = os.environ.get('LOG_LEVEL', 'INFO').upper()
log_level = getattr(logging, log_level_str, logging.INFO)

logging.basicConfig(level=log_level, stream=sys.stdout)
logger = logging.getLogger('myagent')

# Development: LOG_LEVEL=DEBUG python agent.py     -> full traces
# Production:  LOG_LEVEL=WARNING python agent.py  -> only warnings/errors
# Default:     LOG_LEVEL not set                  -> INFO level

logger.debug('This only appears in DEBUG mode')
logger.info('This appears in INFO and DEBUG modes')
logger.warning('This always appears')

اختبار المعرفة: تسجيل التتبع

مراجعة: تسجيل تتبع خطوات الوكيل