AI Agent Observability: คู่มือ Logging และ Tracing ฉบับสมบูรณ์

การสร้าง AI agent ที่ทำงานได้ดีเป็นเรื่องหนึ่ง แต่การเข้าใจว่า agent ทำงานอย่างไรภายในเป็นอีกเรื่องที่สำคัญไม่แพ้กัน ในบทความนี้เราจะพาคุณไปรู้จักกับ AI agent observability ซึ่งครอบคลุม logging และ tracing อย่างละเอียด พร้อมตัวอย่างโค้ดจริงจากเคสการใช้งานของนักพัฒนาอิสระอย่างผมเอง

ทำไมต้อง Observability สำหรับ AI Agent?

ในฐานะนักพัฒนาที่เคย deploy AI agent หลายตัว ผมเจอปัญหาซ้ำๆ ว่า "agent ตอบผิด ทำไม่ได้ แต่ไม่รู้ว่าตรงไหนพัง" ปัญหานี้เกิดจากการขาด observability ในระบบ AI agent

Observability คือความสามารถในการมองเห็นสิ่งที่เกิดขึ้นภายในระบบ แบ่งเป็น 3 เสาหลัก:

• Logging — การบันทึกเหตุการณ์ต่างๆ เป็นข้อความ
• Tracing — การติดตาม request ที่ไหลผ่านระบบ
• Metrics — การวัดค่าตัวเลข เช่น latency, token usage

กรณีศึกษา: นักพัฒนาอิสระ — ระบบ Customer Support Agent

ผมเคยพัฒนา customer support agent สำหรับร้านค้าออนไลน์ ใช้ HolySheep AI เป็น LLM backend ด้วยราคาที่ประหยัดมาก (DeepSeek V3.2 เพียง $0.42/MTok) ทำให้ทดลองได้สบายกระเป๋า

ปัญหาที่เจอคือ agent บางครั้งตอบเรื่องส่งสินค้าผิด ต้อง trace ไป才พบว่า tool "check_shipping_status" ถูกเรียกด้วย parameter ผิด นี่คือประโยชน์ของ tracing ครับ

การ Implement Logging พื้นฐาน

เริ่มจากระบบ logging ที่ดี ผมใช้โครงสร้าง log แบบ structured พร้อม context ที่จำเป็น:

import json
import logging
from datetime import datetime
from typing import Any, Optional
import httpx

Setup logger
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s | %(levelname)s | %(name)s | %(message)s'
)
logger = logging.getLogger("ai_agent")

class AIAgentLogger:
    """Logger สำหรับ AI Agent พร้อม structured output"""
    
    def __init__(self, agent_name: str):
        self.agent_name = agent_name
        self.session_id = self._generate_session_id()
    
    def _generate_session_id(self) -> str:
        return f"{datetime.now().strftime('%Y%m%d%H%M%S')}"
    
    def log_request(
        self, 
        prompt: str, 
        model: str,
        token_count: Optional[int] = None
    ):
        logger.info(json.dumps({
            "event": "llm_request",
            "session": self.session_id,
            "agent": self.agent_name,
            "model": model,
            "prompt_length": len(prompt),
            "prompt_preview": prompt[:100] + "..." if len(prompt) > 100 else prompt,
            "estimated_tokens": token_count
        }))
    
    def log_response(
        self, 
        response: str, 
        latency_ms: float,
        tokens_used: Optional[int] = None
    ):
        logger.info(json.dumps({
            "event": "llm_response",
            "session": self.session_id,
            "agent": self.agent_name,
            "response_length": len(response),
            "response_preview": response[:200] + "..." if len(response) > 200 else response,
            "latency_ms": round(latency_ms, 2),
            "tokens_used": tokens_used
        }))
    
    def log_tool_call(
        self, 
        tool_name: str, 
        parameters: dict[str, Any],
        result: Any,
        success: bool
    ):
        logger.info(json.dumps({
            "event": "tool_call",
            "session": self.session_id,
            "agent": self.agent_name,
            "tool": tool_name,
            "parameters": parameters,
            "success": success,
            "result_preview": str(result)[:150] + "..." if len(str(result)) > 150 else str(result)
        }))
    
    def log_error(self, error: Exception, context: dict[str, Any] = None):
        logger.error(json.dumps({
            "event": "error",
            "session": self.session_id,
            "agent": self.agent_name,
            "error_type": type(error).__name__,
            "error_message": str(error),
            "context": context or {}
        }))

วิธีใช้งาน
agent_logger = AIAgentLogger("customer_support_v2")

ตัวอย่างการใช้งาน
agent_logger.log_request(
    prompt="ลูกค้าถามเรื่องเลขติดตามสินค้า",
    model="deepseek-chat",
    token_count=150
)

จากโค้ดข้างบน ผมออกแบบให้ log ทุก event เป็น JSON format ซึ่งส่งไปยัง ELK stack หรือ Datadog ได้ง่าย สิ่งสำคัญคือการเก็บ session_id เพื่อ trace การทำงานของ request เดียวได้ตลอดทาง

การ Implement Tracing สำหรับ Multi-Step Agent

AI agent ที่ซับซ้อนมักมีหลายขั้นตอน ต้อง trace การไหลของข้อมูลระหว่างขั้นตอน:

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any
import time

class StepStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    SUCCESS = "success"
    FAILED = "failed"

@dataclass
class TraceSpan:
    """Span สำหรับ trace แต่ละขั้นตอน"""
    name: str
    trace_id: str
    span_id: str
    parent_span_id: str = None
    start_time: float = field(default_factory=time.time)
    end_time: float = None
    status: StepStatus = StepStatus.PENDING
    input_data: dict = field(default_factory=dict)
    output_data: Any = None
    error: str = None
    
    def complete(self, output: Any, status: StepStatus = StepStatus.SUCCESS):
        self.end_time = time.time()
        self.output_data = output
        self.status = status
    
    def fail(self, error: Exception):
        self.end_time = time.time()
        self.status = StepStatus.FAILED
        self.error = str(error)
    
    @property
    def duration_ms(self) -> float:
        if self.end_time:
            return (self.end_time - self.start_time) * 1000
        return (time.time() - self.start_time) * 1000
    
    def to_dict(self) -> dict:
        return {
            "trace_id": self.trace_id,
            "span_id": self.span_id,
            "parent_span_id": self.parent_span_id,
            "name": self.name,
            "start_time": self.start_time,
            "end_time": self.end_time,
            "duration_ms": round(self.duration_ms, 2),
            "status": self.status.value,
            "input": self.input_data,
            "output": str(self.output_data)[:500] if self.output_data else None,
            "error": self.error
        }

class DistributedTracer:
    """Tracer สำหรับ track การทำงานแบบ distributed"""
    
    def __init__(self, agent_name: str):
        self.agent_name = agent_name
        self.spans: list[TraceSpan] = []
        self._span_counter = 0
    
    def create_trace_id(self) -> str:
        return f"trace_{datetime.now().strftime('%Y%m%d%H%M%S%f')}"
    
    def create_span_id(self) -> str:
        self._span_counter += 1
        return f"span_{self._span_counter:04d}"
    
    def start_span(
        self, 
        name: str, 
        trace_id: str = None,
        parent_span_id: str = None,
        input_data: dict = None
    ) -> TraceSpan:
        span = TraceSpan(
            name=name,
            trace_id=trace_id or self.create_trace_id(),
            span_id=self.create_span_id(),
            parent_span_id=parent_span_id,
            input_data=input_data or {}
        )
        span.status = StepStatus.RUNNING
        self.spans.append(span)
        return span
    
    def end_trace(self) -> list[dict]:
        """ส่งออก trace ทั้งหมดเป็น dict"""
        return [span.to_dict() for span in self.spans]

ตัวอย่างการใช้งาน
async def handle_customer_inquiry(tracer: DistributedTracer):
    # 1. เริ่ม trace
    root_span = tracer.start_span(
        name="customer_inquiry",
        input_data={"customer_id": "CUST001"}
    )
    
    try:
        # 2. ขั้นตอน: วิเคราะห์คำถาม
        analyze_span = tracer.start_span(
            name="analyze_intent",
            trace_id=root_span.trace_id,
            parent_span_id=root_span.span_id,
            input_data={"query": "พัสดุอยู่ไหน"}
        )
        intent = await analyze_intent("พัสดุอยู่ไหน")
        analyze_span.complete(intent)
        
        # 3. ขั้นตอน: ดึงข้อมูลจาก tool
        tool_span = tracer.start_span(
            name="check_shipping",
            trace_id=root_span.trace_id,
            parent_span_id=analyze_span.span_id,
            input_data={"order_id": "ORD12345"}
        )
        shipping_data = await check_shipping_api("ORD12345")
        tool_span.complete(shipping_data)
        
        # 4. ขั้นตอน: สร้างคำตอบ
        response_span = tracer.start_span(
            name="generate_response",
            trace_id=root_span.trace_id,
            parent_span_id=tool_span.span_id
        )
        response = await generate_response(intent, shipping_data)
        response_span.complete(response)
        
        root_span.complete(response)
        
    except Exception as e:
        root_span.fail(e)
        raise
    
    # Export trace
    trace_data = tracer.end_trace()
    print(f"Trace completed: {len(trace_data)} spans")
    return trace_data

จากโค้ด tracing ผมเห็นชัดเจนว่า intent → shipping check → response generation ทำงานเรียบร้อย ถ้าเกิด error จะเห็นว่าตรงไหน fail และใช้เวลาเท่าไหร่ ซึ่งช่วย debug ได้เร็วมาก

Integration กับ HolySheep AI พร้อม Observability

นี่คือตัวอย่าง complete ที่ผมใช้จริงใน production:

import os
import json
import time
import httpx
from typing import Optional
from dotenv import load_dotenv

load_dotenv()

class HolySheepAIClient:
    """HolySheep AI Client พร้อม built-in logging และ metrics"""
    
    BASE_URL = "https://api.holysheep.ai/v1"
    
    def __init__(self, api_key: str = None):
        self.api_key = api_key or os.getenv("HOLYSHEEP_API_KEY")
        self.logger = AIAgentLogger("holysheep_client")
        self.metrics = {
            "total_requests": 0,
            "total_tokens": 0,
            "total_latency_ms": 0,
            "errors": 0
        }
    
    async def chat_completion(
        self,
        messages: list[dict],
        model: str = "deepseek-chat",
        temperature: float = 0.7,
        max_tokens: int = 2048,
        trace_id: str = None
    ) -> dict:
        """
        ส่ง request ไป HolySheep AI พร้อม track metrics
        ราคา DeepSeek V3.2: $0.42/MTok (ประหยัด 85%+)
        """
        self.metrics["total_requests"] += 1
        start_time = time.time()
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens
        }
        
        self.logger.log_request(
            prompt=json.dumps(messages),
            model=model,
            token_count=max_tokens
        )
        
        try:
            async with httpx.AsyncClient(timeout=30.0) as client:
                response = await client.post(
                    f"{self.BASE_URL}/chat/completions",
                    headers=headers,
                    json=payload
                )
                response.raise_for_status()
                result = response.json()
                
                latency_ms = (time.time() - start_time) * 1000
                usage = result.get("usage", {})
                tokens_used = usage.get("total_tokens", 0)
                
                # Update metrics
                self.metrics["total_tokens"] += tokens_used
                self.metrics["total_latency_ms"] += latency_ms
                
                self.logger.log_response(
                    response=result["choices"][0]["message"]["content"],
                    latency_ms=latency_ms,
                    tokens_used=tokens_used
                )
                
                return {
                    "content": result["choices"][0]["message"]["content"],
                    "usage": usage,
                    "latency_ms": round(latency_ms, 2),
                    "trace_id": trace_id
                }
                
        except httpx.HTTPStatusError as e:
            self.metrics["errors"] += 1
            self.logger.log_error(e, {"status_code": e.response.status_code})
            raise
        except Exception as e:
            self.metrics["errors"] += 1
            self.logger.log_error(e)
            raise
    
    def get_metrics(self) -> dict:
        """ดึง metrics สรุป"""
        avg_latency = (
            self.metrics["total_latency_ms"] / self.metrics["total_requests"]
            if self.metrics["total_requests"] > 0 else 0
        )
        
        # คำนวณค่าใช้จ่าย (ตัวอย่าง: DeepSeek V3.2)
        cost_usd = self.metrics["total_tokens"] / 1_000_000 * 0.42
        
        return {
            **self.metrics,
            "avg_latency_ms": round(avg_latency, 2),
            "estimated_cost_usd": round(cost_usd, 4)
        }

วิธีใช้งาน
async def main():
    client = HolySheepAIClient(api_key="YOUR_HOLYSHEEP_API_KEY")
    
    messages = [
        {"role": "system", "content": "คุณเป็นผู้ช่วยตอบคำถามลูกค้า"},
        {"role": "user", "content": "สินค้าสั่งซื้อเมื่อวานยังไม่ถึง"}
    ]
    
    result = await client.chat_completion(
        messages=messages,
        model="deepseek-chat",
        trace_id="ORD_INQ_001"
    )
    
    print(f"Response: {result['content']}")
    print(f"Latency: {result['latency_ms']}ms")
    print(f"Metrics: {client.get_metrics()}")

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

จุดเด่นของ integration นี้คือ:

• latency เฉลี่ย <50ms ตามที่ HolySheep รับประกัน
• ทุก request มี trace_id เพื่อติดตาม
• คำนวณค่าใช้จ่ายอัตโนมัติจาก token usage
• เก็บ metrics สำหรับวิเคราะห์ภายหลัง

Visualization และ Monitoring Dashboard

เมื่อมีข้อมูล log และ trace แล้ว ควร visualize ด้วย dashboard เพื่อเห็นภาพรวม:

from dataclasses import dataclass
from typing import Callable, Any
import time

@dataclass
class MonitoringDashboard:
    """Simple monitoring สำหรับ AI Agent"""
    
    def __init__(self):
        self.sla_thresholds = {
            "latency_p95_ms": 500,
            "error_rate_percent": 1.0,
            "cost_per_request_usd": 0.01
        }
        self.alerts: list[dict] = []
    
    def check_sla(self, metrics: dict) -> list[str]:
        """ตรวจสอบ SLA และส่ง alert ถ้าผิดเกณฑ์"""
        issues = []
        
        if metrics.get("avg_latency_ms", 0) > self.sla_thresholds["latency_p95_ms"]:
            issues.append(f"⚠️ Latency สูงเกิน SLA: {metrics['avg_latency_ms']}ms")
        
        if metrics.get("errors", 0) > 0:
            error_rate = (metrics["errors"] / metrics["total_requests"] * 100) if metrics["total_requests"] > 0 else 0
            if error_rate > self.sla_thresholds["error_rate_percent"]:
                issues.append(f"⚠️ Error rate สูง: {error_rate:.2f}%")
        
        avg_cost = metrics.get("estimated_cost_usd", 0) / max(metrics.get("total_requests", 1), 1)
        if avg_cost > self.sla_thresholds["cost_per_request_usd"]:
            issues.append(f"⚠️ ค่าใช้จ่ายสูงต่อ request: ${avg_cost:.4f}")
        
        return issues
    
    def generate_report(self, metrics: dict) -> str:
        """สร้างรายงานสรุป"""
        issues = self.check_sla(metrics)
        
        report = f"""
╔══════════════════════════════════════════════════════════╗
║            AI Agent Monitoring Report                     ║
╠══════════════════════════════════════════════════════════╣
║ Total Requests:      {metrics.get('total_requests', 0):>10}                    ║
║ Total Tokens:        {metrics.get('total_tokens', 0):>10}                    ║
║ Avg Latency:         {metrics.get('avg_latency_ms', 0):>10.2f} ms                 ║
║ Error Count:         {metrics.get('errors', 0):>10}                    ║
║ Estimated Cost:      ${metrics.get('estimated_cost_usd', 0):>10.4f}                 ║
╠══════════════════════════════════════════════════════════╣"""
        
        if issues:
            report += "\n║ ALERTS:                                                  ║"
            for issue in issues:
                report += f"\n║ {issue:<54}║"
        else:
            report += "\n║ ✅ All SLA metrics OK                                    ║"
        
        report += "\n╚══════════════════════════════════════════════════════════╝"
        return report

ใช้งาน
dashboard = MonitoringDashboard()
metrics = client.get_metrics()
print(dashboard.generate_report(metrics))

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

1. Logging ไม่ครบ flow — เห็น error แต่ไม่รู้ context

ปัญหา: log มีแค่ error message แต่ไม่มี prompt, parameters หรือ trace_id ทำให้ debug ยาก

วิธีแก้: เพิ่ม structured logging พร้อม context ทุก event

# ❌ วิธีผิด - log แค่ error
logger.error("API call failed")

✅ วิธีถูก - log พร้อม context เต็มรูปแบบ
logger.error(json.dumps({
    "event": "api_call_failed",
    "trace_id": current_trace_id,
    "span_id": current_span_id,
    "error_type": type(e).__name__,
    "error_message": str(e),
    "request_params": {"url": url, "method": method},
    "retry_count": retry_count,
    "timestamp": datetime.now().isoformat()
}))

2. Trace ID ไม่ตรงกันระหว่าง service — ไม่สามารถ track request

ปัญหา: เรียก service A สร้าง trace_id ใหม่ แล้วเรียก service B ก็สร้าง trace_id ใหม่อีก ทำให้ไม่ต่อกัน

วิธีแก้: Pass trace_id ผ่าน HTTP headers หรือ context object

# ❌ วิธีผิด - สร้าง trace_id ใหม่ทุก service
async def call_shipping_service(order_id):
    tracer = DistributedTracer("shipping")  # trace_id ใหม่!
    ...

✅ วิธีถูก - ใช้ trace_id เดิมจาก parent
async def call_shipping_service(order_id, trace_id, parent_span_id):
    tracer = DistributedTracer("shipping")
    span = tracer.start_span(
        name="shipping_check",
        trace_id=trace_id,  # ใช้ trace_id จาก parent
        parent_span_id=parent_span_id
    )
    ...

เรียกใช้โดยส่ง trace context ต่อๆ กัน
await call_shipping_service(
    order_id="ORD123",
    trace_id=root_span.trace_id,
    parent_span_id=root_span.span_id
)

3. เก็บ token usage ไม่ถูกต้อง — คำนวณค่าใช้จ่ายผิด

ปัญหา: ใช้ max_tokens แทน token ที่ใช้จริง ทำให้ประมาณค่าใช้จ่ายสูงเกินจริง

วิธีแก้: ใช้ค่าจาก response.usage ที่ API ส่งกลับมาเสมอ

# ❌ วิธีผิด - ใช้ max_tokens ที่ส่งไป
tokens_used = max_tokens  # ผิด! นี่คือ limit ไม่ใช่ usage

✅ วิธีถูก - ใช้ usage จาก response
result = await client.chat_completion(messages)
tokens_used = result["usage"]["total_tokens"]

ตรวจสอบ response structure จาก HolySheep
{
  "usage": {
    "prompt_tokens": 150,
    "completion_tokens": 200,
    "total_tokens": 350
  }
}
cost = tokens_used / 1_000_000 * 0.42  # DeepSeek V3.2: $0.42/MTok

4. Metric aggregation ผิด — เห็นตัวเลขสถิติไม่ถูกต้อง

ปัญหา: ใช้ arithmetic mean กับ latency ที่มี outlier ทำให้เห็น latency ต่ำเกินจริง

วิธีแก้: ใช้ percentile เช่น p50, p95, p99

from statistics import mean, median

❌ วิธีผิด - ใช้ mean
avg_latency = mean(latencies)  # outlier ทำให้สูงเกินจริง

✅ วิธีถูก - ใช้ percentile + median
def calculate_latency_metrics(latencies: list[float]) -> dict:
    sorted_latencies = sorted(latencies)
    n = len(sorted_latencies)
    
    return {
        "p50": sorted_latencies[int(n * 0.50)] if n > 0 else 0,
        "p95": sorted_latencies[int(n * 0.95)] if n > 0 else 0,
        "p99": sorted_latencies[int(n * 0.99)] if n > 0 else 0,
        "median": median(latencies),
        "mean": mean(latencies)
    }

ตัวอย่างผลลัพธ์
{"p50": 45.2, "p95": 120.5, "p99": 350.0, "median": 45.2, "mean": 78.3}
p95 สำคัญกว่า mean สำหรับ SLA

สรุป

AI agent observability ไม่ใช่ luxury แต่เป็น ความจำเป็น สำหรับ production system การลงทุนเวลาสร้างระบบ logging และ tracing ที่ดีจะช่วยประหยัดเวลาการ debug ได้มหาศาล

จากประสบการณ์ของผม การใช้ HolySheep AI ร่วมกับ observability ที่ดี ช่วยให้:

แหล่งข้อมูลที่เกี่ยวข้อง

• 📚 บทช่วยสอน AI API
• 💰 ดูราคา
• 📖 เอกสารสำหรับนักพัฒนา
• 🚀 สมัครฟรี

บทความที่เกี่ยวข้อง

• LangChain RAG Implementation: Vector Store และ Retriever Set
• OpenAI vs Anthropic API: กลยุทธ์ความแตกต่างและวิธีเลือกที่เห
• Dify 模板实战：构建企业级邮件回复工作流