ในโลกของ Multi-Agent Orchestration การ debug agent ที่ทำงานพร้อมกันหลายตัวนั้นยากกว่าการ debug โค้ดปกติมาก เนื่องจากมี race condition, state management และ communication pattern ที่ซับซ้อน ในบทความนี้ผมจะแชร์เทคนิคที่ใช้จริงในการ visualize agent behavior รวมถึงการตั้งค่า monitoring ที่ช่วยให้เห็น flow ของข้อมูลระหว่าง agent อย่างชัดเจน
สถาปัตยกรรม Debugging System สำหรับ CrewAI
ก่อนจะลงลึกเรื่อง code มาทำความเข้าใจ architecture ของ debugging system ที่เราจะสร้างกัน
┌─────────────────────────────────────────────────────────────────┐
│ CrewAI Debug Architecture │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Agent 1 │───▶│ Agent 2 │───▶│ Agent 3 │───▶│ Output │ │
│ │ (Research)│ │(Analyzer)│ │(Reporter)│ │(Summary) │ │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ └──────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ Event Tracer (Structured Logging) │ │
│ │ - Timestamp (ms precision) │ │
│ │ - Agent ID & Task ID │ │
│ │ - Input/Output tokens │ │
│ │ - Latency per operation │ │
│ │ - Tool calls & results │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ Visualization Dashboard (Real-time) │ │
│ │ - Sequence diagram │ │
│ │ - Token usage graph │ │
│ │ - Error distribution │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
การตั้งค่า Event Tracer พร้อม HolySheep AI
สำหรับการ monitor agent behavior ใน production ผมแนะนำให้ใช้ HolySheep AI เพราะมี latency เฉลี่ยต่ำกว่า 50ms และราคาถูกกว่าคู่แข่งถึง 85% ทำให้เหมาะสำหรับงานที่ต้อง trace หลายพัน events ต่อวัน
import json
import time
import asyncio
from datetime import datetime
from typing import Any, Dict, List, Optional
from dataclasses import dataclass, field, asdict
from enum import Enum
import logging
Configure structured logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s.%(msecs)03d | %(levelname)s | %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
logger = logging.getLogger("crewai-debugger")
class EventType(Enum):
AGENT_START = "agent_start"
AGENT_END = "agent_end"
TOOL_CALL = "tool_call"
TOOL_RESULT = "tool_result"
TASK_ASSIGN = "task_assign"
ERROR = "error"
LLM_REQUEST = "llm_request"
LLM_RESPONSE = "llm_response"
@dataclass
class AgentEvent:
"""Structured event สำหรับ tracing agent behavior"""
timestamp: float = field(default_factory=time.time)
event_type: str
agent_id: str
task_id: str
session_id: str
duration_ms: Optional[float] = None
input_tokens: Optional[int] = None
output_tokens: Optional[int] = None
total_tokens: Optional[int] = None
model: Optional[str] = None
latency_ms: Optional[float] = None
error: Optional[str] = None
metadata: Dict[str, Any] = field(default_factory=dict)
def to_json(self) -> str:
return json.dumps(asdict(self), ensure_ascii=False)
class CrewAIDebugger:
"""Main debugger class สำหรับ CrewAI behavior visualization"""
def __init__(self, session_id: str, output_dir: str = "./debug_logs"):
self.session_id = session_id
self.output_dir = output_dir
self.events: List[AgentEvent] = []
self.start_time = time.time()
self._lock = asyncio.Lock()
# Create timestamped log file
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
self.log_file = f"{output_dir}/crewai_trace_{session_id}_{timestamp}.jsonl"
logger.info(f"🔍 Debugger initialized | Session: {session_id}")
async def log_event(self, event: AgentEvent):
"""Async-safe event logging with file flush"""
async with self._lock:
self.events.append(event)
with open(self.log_file, 'a', encoding='utf-8') as f:
f.write(event.to_json() + '\n')
# Real-time logging
logger.info(
f"[{event.event_type.upper()}] "
f"Agent: {event.agent_id} | "
f"Duration: {event.duration_ms:.2f}ms | "
f"Tokens: {event.total_tokens or 'N/A'}"
)
async def trace_agent_execution(
self,
agent_id: str,
task_id: str,
coro
) -> Any:
"""Context manager สำหรับ trace agent execution"""
start_time = time.time()
# Log start event
start_event = AgentEvent(
event_type=EventType.AGENT_START.value,
agent_id=agent_id,
task_id=task_id,
session_id=self.session_id
)
await self.log_event(start_event)
try:
# Execute the agent coroutine
result = await coro
# Log success event
end_time = time.time()
duration_ms = (end_time - start_time) * 1000
end_event = AgentEvent(
event_type=EventType.AGENT_END.value,
agent_id=agent_id,
task_id=task_id,
session_id=self.session_id,
duration_ms=duration_ms
)
await self.log_event(end_event)
return result
except Exception as e:
# Log error event
error_event = AgentEvent(
event_type=EventType.ERROR.value,
agent_id=agent_id,
task_id=task_id,
session_id=self.session_id,
error=str(e),
metadata={"exception_type": type(e).__name__}
)
await self.log_event(error_event)
raise
async def log_llm_call(
self,
agent_id: str,
task_id: str,
model: str,
input_tokens: int,
output_tokens: int,
latency_ms: float
):
"""Log LLM API call พร้อม token usage และ latency"""
event = AgentEvent(
event_type=EventType.LLM_RESPONSE.value,
agent_id=agent_id,
task_id=task_id,
session_id=self.session_id,
model=model,
input_tokens=input_tokens,
output_tokens=output_tokens,
total_tokens=input_tokens + output_tokens,
latency_ms=latency_ms
)
await self.log_event(event)
Initialize global debugger instance
debugger = CrewAIDebugger(session_id="production-001")
การ Integrate กับ HolySheep AI API
สำหรับการ call LLM ใน production ผมใช้ HolySheep AI เพราะมีราคาที่คุ้มค่ามาก โดยเฉพาะ DeepSeek V3.2 ที่ราคาเพียง $0.42/MTok ซึ่งเหมาะสำหรับ trace logs ที่ต้องส่งไปเก็บ
import os
from openai import AsyncOpenAI
from litellm import acompletion
from typing import Optional
HolySheep AI Configuration
ราคา: GPT-4.1 $8/MTok, Claude Sonnet 4.5 $15/MTok, DeepSeek V3.2 $0.42/MTok
ใช้ ¥1=$1 rate ประหยัด 85%+ จากราคาเดิม
HOLYSHEEP_API_KEY = os.environ.get("YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
Initialize HolySheep client
holysheep_client = AsyncOpenAI(
api_key=HOLYSHEEP_API_KEY,
base_url=HOLYSHEEP_BASE_URL,
timeout=30.0,
max_retries=3
)
class CostTracker:
"""Track token usage และคำนวณค่าใช้จ่าย"""
# HolySheep AI Pricing (2026)
PRICING = {
"gpt-4.1": {"input": 2.0, "output": 8.0}, # $2/$8 per MTok
"gpt-4.1-mini": {"input": 0.15, "output": 0.6}, # $0.15/$0.6 per MTok
"claude-sonnet-4.5": {"input": 3.0, "output": 15.0}, # $3/$15 per MTok
"gemini-2.5-flash": {"input": 0.125, "output": 0.50}, # $0.125/$0.50 per MTok
"deepseek-v3.2": {"input": 0.14, "output": 0.42}, # $0.14/$0.42 per MTok
}
def __init__(self):
self.total_input_tokens = 0
self.total_output_tokens = 0
self.costs_by_model: dict[str, float] = {}
def add_usage(self, model: str, input_tokens: int, output_tokens: int):
self.total_input_tokens += input_tokens
self.total_output_tokens += output_tokens
if model not in self.PRICING:
return
input_cost = (input_tokens / 1_000_000) * self.PRICING[model]["input"]
output_cost = (output_tokens / 1_000_000) * self.PRICING[model]["output"]
total_cost = input_cost + output_cost
self.costs_by_model[model] = self.costs_by_model.get(model, 0) + total_cost
def get_summary(self) -> dict:
return {
"total_input_tokens": self.total_input_tokens,
"total_output_tokens": self.total_output_tokens,
"total_cost_usd": sum(self.costs_by_model.values()),
"cost_breakdown": self.costs_by_model
}
cost_tracker = CostTracker()
async def call_llm_with_tracing(
agent_id: str,
task_id: str,
messages: list,
model: str = "deepseek-v3.2",
temperature: float = 0.7,
max_tokens: int = 2048
) -> tuple[str, float]:
"""
Call LLM ผ่าน HolySheep พร้อม trace latency และ token usage
Returns: (response_content, latency_ms)
"""
start_time = time.time()
try:
response = await holysheep_client.chat.completions.create(
model=model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens,
stream=False
)
end_time = time.time()
latency_ms = (end_time - start_time) * 1000
# Extract token usage
usage = response.usage
input_tokens = usage.prompt_tokens
output_tokens = usage.completion_tokens
total_tokens = usage.total_tokens
# Track costs
cost_tracker.add_usage(model, input_tokens, output_tokens)
# Log to debugger
await debugger.log_llm_call(
agent_id=agent_id,
task_id=task_id,
model=model,
input_tokens=input_tokens,
output_tokens=output_tokens,
latency_ms=latency_ms
)
content = response.choices[0].message.content
logger.info(
f"[LLM] {model} | "
f"Latency: {latency_ms:.2f}ms | "
f"Tokens: {total_tokens:,} (in:{input_tokens} out:{output_tokens})"
)
return content, latency_ms
except Exception as e:
logger.error(f"[LLM ERROR] {agent_id}/{task_id}: {str(e)}")
raise
async def test_holy_sheep_connection():
"""Test HolySheep API connection พร้อม benchmark"""
test_messages = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Hello, respond with a short greeting."}
]
# Test DeepSeek V3.2 (cheapest option)
print("🧪 Testing HolySheep AI Connection...")
print("-" * 50)
result, latency = await call_llm_with_tracing(
agent_id="test-agent",
task_id="connection-test",
messages=test_messages,
model="deepseek-v3.2"
)
print(f"✅ Response: {result[:100]}...")
print(f"⏱️ Latency: {latency:.2f}ms")
print(f"📊 Cost Summary: {cost_tracker.get_summary()}")
print("-" * 50)
# Verify latency is under 50ms as advertised
assert latency < 200, f"Latency too high: {latency:.2f}ms"
print(f"✅ Latency verification passed: {latency:.2f}ms < 200ms")
Run test
if __name__ == "__main__":
asyncio.run(test_holy_sheep_connection())
Real-time Visualization Dashboard
หลังจากมี structured logs แล้ว ต่อไปจะสร้าง dashboard สำหรับ visualize agent flow แบบ real-time
import asyncio
import json
from collections import defaultdict
from datetime import datetime, timedelta
from typing import Dict, List
class AgentVisualizer:
"""Visualize CrewAI agent behavior เป็น sequence diagram"""
def __init__(self):
self.events: List[AgentEvent] = []
self.agent_timelines: Dict[str, List[Dict]] = defaultdict(list)
self.error_count = 0
self.total_duration_ms = 0
def load_from_file(self, log_file: str):
"""Load events จาก JSONL file"""
with open(log_file, 'r', encoding='utf-8') as f:
for line in f:
data = json.loads(line)
event = AgentEvent(**data)
self.events.append(event)
self._process_event(event)
def _process_event(self, event: AgentEvent):
"""Process event เพื่อสร้าง timeline"""
if event.event_type == EventType.ERROR.value:
self.error_count += 1
self.agent_timelines[event.agent_id].append({
"timestamp": event.timestamp,
"type": event.event_type,
"duration_ms": event.duration_ms,
"tokens": event.total_tokens,
"error": event.error
})
def generate_sequence_diagram(self) -> str:
"""Generate Mermaid sequence diagram"""
lines = ["```mermaid", "sequenceDiagram"]
# Sort events by timestamp
sorted_events = sorted(self.events, key=lambda x: x.timestamp)
agent_sequence = []
for event in sorted_events:
if event.agent_id not in agent_sequence:
agent_sequence.append(event.agent_id)
lines.append(f" participant {event.agent_id}")
lines.append("")
# Generate message flows
for i in range(len(sorted_events) - 1):
curr = sorted_events[i]
next_e = sorted_events[i + 1]
if curr.event_type == EventType.AGENT_END.value:
lines.append(
f" {curr.agent_id}->>+{next_e.agent_id}: "
f"Forward task {curr.task_id}"
)
lines.append("```")
return "\n".join(lines)
def generate_metrics_report(self) -> Dict:
"""Generate performance metrics report"""
# Group by agent
agent_stats = {}
for agent_id, timeline in self.agent_timelines.items():
durations = [e["duration_ms"] for e in timeline if e["duration_ms"]]
token_totals = [e["tokens"] for e in timeline if e["tokens"]]
agent_stats[agent_id] = {
"execution_count": len(timeline),
"avg_duration_ms": sum(durations) / len(durations) if durations else 0,
"min_duration_ms": min(durations) if durations else 0,
"max_duration_ms": max(durations) if durations else 0,
"total_tokens": sum(token_totals),
"error_count": sum(1 for e in timeline if e["error"])
}
return {
"session_id": self.events[0].session_id if self.events else "N/A",
"total_events": len(self.events),
"total_errors": self.error_count,
"agent_stats": agent_stats,
"cost_summary": cost_tracker.get_summary()
}
def print_timeline(self):
"""Print human-readable timeline"""
print("\n" + "=" * 80)
print("📊 CREWAI EXECUTION TIMELINE")
print("=" * 80)
sorted_events = sorted(self.events, key=lambda x: x.timestamp)
for event in sorted_events:
ts = datetime.fromtimestamp(event.timestamp).strftime("%H:%M:%S.%f")[:-3]
if event.event_type == EventType.AGENT_START.value:
icon = "🚀"
status = "STARTED"
elif event.event_type == EventType.AGENT_END.value:
icon = "✅"
status = f"ENDED ({event.duration_ms:.2f}ms)"
elif event.event_type == EventType.LLM_RESPONSE.value:
icon = "💬"
status = f"LLM ({event.latency_ms:.2f}ms, {event.total_tokens} tokens)"
elif event.event_type == EventType.ERROR.value:
icon = "❌"
status = f"ERROR: {event.error}"
else:
icon = "📝"
status = event.event_type
print(f"{ts} | {icon} [{status:30}] | {event.agent_id} | Task: {event.task_id}")
print("=" * 80)
# Print metrics
metrics = self.generate_metrics_report()
print("\n📈 PERFORMANCE METRICS")
print("-" * 40)
print(f"Total Events: {metrics['total_events']}")
print(f"Total Errors: {metrics['total_errors']}")
print("\n👥 AGENT BREAKDOWN")
for agent_id, stats in metrics['agent_stats'].items():
print(f"\n Agent: {agent_id}")
print(f" Executions: {stats['execution_count']}")
print(f" Avg Duration: {stats['avg_duration_ms']:.2f}ms")
print(f" Total Tokens: {stats['total_tokens']:,}")
print(f" Errors: {stats['error_count']}")
print("\n💰 COST BREAKDOWN")
cost = metrics['cost_summary']
print(f" Input Tokens: {cost['total_input_tokens']:,}")
print(f" Output Tokens: {cost['total_output_tokens']:,}")
print(f" Total Cost: ${cost['total_cost_usd']:.4f}")
for model, cost_val in cost['cost_breakdown'].items():
print(f" - {model}: ${cost_val:.4f}")
print("\n" + "=" * 80)
Demo usage
async def demo_visualization():
"""Demo visualization with sample data"""
visualizer = AgentVisualizer()
# Generate sample timeline
base_time = time.time()
sample_events = [
AgentEvent(timestamp=base_time, event_type="agent_start", agent_id="researcher", task_id="t1", session_id="demo"),
AgentEvent(timestamp=base_time+0.1, event_type="llm_request", agent_id="researcher", task_id="t1", session_id="demo", model="deepseek-v3.2", latency_ms=45.2, total_tokens=1200),
AgentEvent(timestamp=base_time+0.15, event_type="agent_end", agent_id="researcher", task_id="t1", session_id="demo", duration_ms=150.5),
AgentEvent(timestamp=base_time+0.16, event_type="agent_start", agent_id="analyzer", task_id="t2", session_id="demo"),
AgentEvent(timestamp=base_time+0.5, event_type="llm_request", agent_id="analyzer", task_id="t2", session_id="demo", model="deepseek-v3.2", latency_ms=48.1, total_tokens=2100),
AgentEvent(timestamp=base_time+0.55, event_type="agent_end", agent_id="analyzer", task_id="t2", session_id="demo", duration_ms=390.2),
]
for event in sample_events:
visualizer.events.append(event)
visualizer._process_event(event)
# Add some costs
cost_tracker.add_usage("deepseek-v3.2", 1200, 800)
cost_tracker.add_usage("deepseek-v3.2", 2100, 1200)
visualizer.print_timeline()
print("\n" + visualizer.generate_sequence_diagram())
if __name__ == "__main__":
asyncio.run(demo_visualization())
Concurrent Agent Execution Debugging
สำหรับ CrewAI ที่รันหลาย agent พร้อมกัน การ debug ยิ่งซับซ้อน เพราะต้อง track race conditions และ deadlocks
import asyncio
from contextlib import asynccontextmanager
from typing import Set
import threading
class ConcurrencyDebugger:
"""Debug concurrent agent execution"""
def __init__(self):
self.active_agents: Set[str] = set()
self.lock = asyncio.Lock()
self.execution_graph: Dict[str, List[str]] = defaultdict(list)
self._thread_safe_lock = threading.Lock()
@asynccontextmanager
async def track_agent(self, agent_id: str, task_id: str):
"""Track agent execution พร้อม deadlock detection"""
acquired = await self.lock.acquire(timeout=10.0)
if not acquired:
error_msg = f"⚠️ DEADLOCK DETECTED: Agent {agent_id} could not acquire lock"
logger.error(error_msg)
# Log deadlock event
await debugger.log_event(AgentEvent(
event_type="deadlock",
agent_id=agent_id,
task_id=task_id,
session_id=debugger.session_id,
error=error_msg
))
raise TimeoutError(error_msg)
try:
self.active_agents.add(agent_id)
logger.info(f"🔒 Agent {agent_id} acquired lock | Active: {self.active_agents}")
await debugger.log_event(AgentEvent(
event_type="lock_acquired",
agent_id=agent_id,
task_id=task_id,
session_id=debugger.session_id,
metadata={"active_agents": list(self.active_agents)}
))
yield
finally:
self.active_agents.discard(agent_id)
self.lock.release()
logger.info(f"🔓 Agent {agent_id} released lock | Active: {self.active_agents}")
async def run_parallel_agents(
self,
agents: List[tuple[str, str, coroutine]]
) -> List[Any]:
"""
Run multiple agents in parallel พร้อม concurrency tracking
Args:
agents: List of (agent_id, task_id, coroutine)
Returns:
List of results from each agent
"""
logger.info(f"🚀 Starting {len(agents)} agents in parallel")
async def run_with_tracking(agent_id: str, task_id: str, coro):
async with self.track_agent(agent_id, task_id):
return await debugger.trace_agent_execution(agent_id, task_id, coro)
# Create tasks
tasks = [
asyncio.create_task(run_with_tracking(aid, tid, coro))
for aid, tid, coro in agents
]
# Wait for all to complete
results = await asyncio.gather(*tasks, return_exceptions=True)
# Process results
processed_results = []
for i, result in enumerate(results):
if isinstance(result, Exception):
logger.error(f"Agent {agents[i][0]} failed: {result}")
processed_results.append({"error": str(result), "agent_id": agents[i][0]})
else:
processed_results.append({"result": result, "agent_id": agents[i][0]})
return processed_results
Usage example
async def demo_concurrent_debugging():
debugger = CrewAIDebugger(session_id="concurrent-test")
concurrency_debugger = ConcurrencyDebugger()
# Define sample agents
async def researcher_agent():
await asyncio.sleep(0.5) # Simulate work
return "Research completed"
async def analyzer_agent():
await asyncio.sleep(0.3) # Simulate work
return "Analysis completed"
async def reporter_agent():
await asyncio.sleep(0.2) # Simulate work
return "Report generated"
# Run concurrently
agents = [
("researcher", "task-1", researcher_agent()),
("analyzer", "task-2", analyzer_agent()),
("reporter", "task-3", reporter_agent()),
]
print("⏳ Running 3 agents concurrently...")
start = time.time()
results = await concurrency_debugger.run_parallel_agents(agents)
elapsed = (time.time() - start) * 1000
print(f"\n✅ All agents completed in {elapsed:.2f}ms")
print(f"📊 Expected: ~500ms (max of all agents)")
for r in results:
print(f" - {r.get('agent_id')}: {'✅' if 'result' in r else '❌'}")
ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
1. JSONL Log File Corruption เมื่อ Process หยุดกลางคัน
อาการ: Log file เสียหายเพราะไม่ได้ flush หรือ process ถูก kill กลางทาง ทำให้อ่านไฟล์ไม่ได้
# ❌ วิธีที่ผิด - ไม่มี proper file handling
async def bad_logging(event: AgentEvent):
with open("log.jsonl", "a") as f:
f.write(event.to_json() + "\n")
# ไม่มี flush - ข้อมูลอาจตกหายถ้า crash
✅ วิธีที่ถูก - ใช้ atomic write กับ temporary file
import tempfile
import os
import fcntl
async def safe_logging(event: AgentEvent, log_file: str):
"""Atomic write พร้อม file locking"""
temp_file = f"{log_file}.tmp"
# Write to temp file
with open(temp_file, 'w', encoding='utf-8') as f:
f.write(event.to_json() + '\n')
f.flush()
os.fsync(f.fileno()) # Force write to disk
# Atomic rename
os.rename(temp_file, log_file)
# Alternative: Use file locking for concurrent writes
with open(log_file, 'a', encoding='utf-8') as f:
fcntl.flock(f.fileno(), fcntl.LOCK_EX)
try:
f.write(event.to_json() + '\n')
f.flush()
finally:
fcntl.flock(f.fileno(), fcntl.LOCK_UN)
2. Memory Leak จาก Event List ที่ไม่ถูก Clear
อาการ: Memory usage เพิ่มขึ้นเรื่อยๆ เมื่อรันนานๆ เพราะ events list ไม่ถูก clear
# ❌ วิธีที่ผิด - events โตเรื่อยๆ
class BadDebugger:
def __init__(self):
self.events = [] # ไม่มีวัน clear
async def log_event(self, event):
self.events.append(event) # Memory leak!
# 1 ล้าน events = ~500MB+ RAM
✅ วิธีที่ถูก - Implement ring buffer และ auto-flush
from collections import deque
class MemoryEfficientDebugger:
MAX_EVENTS_IN_MEMORY = 10000
def __init__(self, log_file: str):
self.events = deque(maxlen=self.MAX_EVENTS_IN_MEMORY) # Ring buffer
self.log_file = log_file
self._flush_count = 0
async def log_event(self, event: AgentEvent):
# Add to ring buffer (auto-evict oldest)
self.events.append(event)
# Write to disk immediately
with open(self.log_file, 'a') as f:
f.write(event.to_json() + '\n')
# Periodic full flush
if len(self.events) >= self.MAX_EVENTS_IN_MEMORY:
await self._flush_to_disk()
async def _flush_to_disk(self):
"""Flush all events to separate analysis file"""
self._flush_count += 1
flush_file = f"{self.log_file}.flush_{self._flush_count}.jsonl"
with open(flush_file, 'w') as f:
for event in self.events:
f.write(event.to_json() + '\n')
self.events.clear() # Free memory
print(f"💾 Flushed {self.MAX_EVENTS_IN_MEMORY} events to {flush_file}")
3. Race Condition ใน Async Lock Acquisition
อาการ: Deadlock หรือ event order สลับกัน โดยเฉพาะเมื่อรันหลาย concurrent tasks
# ❌ วิธีที่ผิด - ใช้ threading.Lock แทน asyncio.Lock
class BadConcurrencyHandler:
def __init__(self):
self.lock = threading.Lock() # Wrong!
async def process(self, agent_id: str):
with self.lock: # Blocking call in async context
await asyncio.sleep(1) # This will block entire event loop!
print(f"Processed {agent_id}")
✅ วิธีที่ถูก - ใช้ asyncio primitives อย่างถูกต้อง
import asyncio
class GoodConcurrencyHandler:
def __init__(self):
self.lock = asyncio.Lock() # Correct!
async def process(self, agent_id: str):
async with self.lock: # Non-blocking async lock
await asyncio.sleep(1) # Yields to event loop
print(f"Processed {agent_id}")
# หรือใช้ Semaphore สำหรับ concurrent limit
def __init__(self, max_concurrent: int = 5):
self.semaphore = asyncio.Semaphore(max_concurrent)
async def process(self, agent_id: str):
async with self.semaphore:
# ไม่เกิน max_concurrent พร้อมกัน
await self._do_work(agent_id)
4. Token Counting ไม่แม่นยำกับ Cached Results
อาการ: ค่าใช้จ่ายไม่ตรงกับ API bill เพราะไม่ได้คิด cached tokens
# ❌ วิธีที่�