Ba tháng trước, đội ngũ kỹ thuật của một sàn thương mại điện tử quy mô 2 triệu người dùng đối mặt với cơn bão: hàng nghìn ticket hỗ trợ mỗi ngày, đội ngũ 50 người không thể xử lý nổi, và chi phí vận hành chatbot AI tăng 300% chỉ trong 6 tháng. Họ cần một giải pháp automation thông minh — và câu trả lời nằm ở AutoGen Code Interpreter Agent.
Bài viết này là bản tổng hợp kinh nghiệm triển khai thực chiến của tôi sau khi cấu hình thành công hệ thống multi-agent cho 3 doanh nghiệp khác nhau. Tôi sẽ hướng dẫn bạn từng bước, kèm code có thể chạy ngay, và đặc biệt — cách tiết kiệm 85%+ chi phí API khi sử dụng HolySheep AI thay vì các provider phương Tây.
Tại Sao Cần AutoGen Code Interpreter Agent?
AutoGen là framework multi-agent được Microsoft phát triển, cho phép nhiều AI agent giao tiếp và cộng tác để giải quyết tác vụ phức tạp. Trong đó, Code Interpreter Agent là agent có khả năng:
- Viết và thực thi code Python trực tiếp
- Phân tích dữ liệu, vẽ biểu đồ, xử lý file
- Tính toán số học phức tạp với độ chính xác cao
- Tự động sửa lỗi (self-correction) khi code chạy thất bại
Kiến Trúc Tổng Quan Hệ Thống
Trước khi code, hãy hiểu rõ kiến trúc mà chúng ta sẽ triển khai:
+------------------+ +-----------------------+ +------------------------+
| User Request | --> | Orchestrator Agent | --> | Code Interpreter Agent |
+------------------+ +-----------------------+ +------------------------+
| |
v v
+----------------+ +-------------------+
| Data Retriever| | Sandbox Python |
+----------------+ +-------------------+
| |
v v
+-------------------------------------------------+
| HolySheep AI API |
| base_url: https://api.holysheep.ai/v1 |
| Models: GPT-4.1, Claude Sonnet, Gemini 2.5 |
+-------------------------------------------------+
Cài Đặt Môi Trường
Đầu tiên, cài đặt các dependencies cần thiết. Tôi khuyên dùng Python 3.10+ cho tính ổn định:
# Cài đặt AutoGen và các dependencies
pip install autogen-agentchat autogen-ext[openai] pyautogen
Package bổ sung cho code execution
pip install jupyter-client matplotlib pandas numpy
Kiểm tra version
python -c "import autogen; print(autogen.__version__)"Code Mẫu Hoàn Chỉnh — Cấu Hình Code Interpreter Agent
Đây là code production-ready mà tôi đã deploy thực tế. Copy và chạy ngay:
import os
import json
from typing import Annotated
from autogen import Agent, ConversableAgent
from autogen.agentchat.conversable_agent import ConversableAgent
from autogen.agentchat.group_chat import GroupChat, GroupChatManager
from autogen.agentchat import AssistMsg
========== CẤU HÌNH HOLYSHEEP AI ==========
QUAN TRỌNG: Không bao giờ hardcode API key trong production
Sử dụng environment variable hoặc secret manager
os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"
os.environ["OPENAI_BASE_URL"] = "https://api.holysheep.ai/v1"
Cấu hình cho Code Interpreter Agent
code_interpreter_config = {
"model": "gpt-4.1", # Model: GPT-4.1 ($8/MTok) hoặc DeepSeek V3.2 ($0.42/MTok)
"price": [0.008, 0.024], # Input/Output price per MTok
"temperature": 0.3, # Độ sáng tạo thấp cho code execution
"timeout": 120, # Timeout 120 giây cho mỗi execution
"max_tokens": 4096,
"seed": 42, # Reproducible results
}
System prompt cho Code Interpreter
CODE_INTERPRETER_SYSTEM_PROMPT = """Bạn là Code Interpreter Agent chuyên nghiệp.
Nhiệm vụ của bạn:
1. Viết code Python chính xác, có error handling
2. Thực thi code an toàn trong sandbox
3. Giải thích kết quả một cách rõ ràng
4. Tự động sửa lỗi nếu code thất bại (tối đa 3 lần retry)
5. Trả về kết quả dạng JSON khi cần thiết
LUÔN LUÔN:
- Import thư viện cần thiết trước khi sử dụng
- Wrap code trong try-except để xử lý lỗi
- Print intermediate results để debug
- Trả về kết quả cuối cùng có cấu trúc rõ ràng
Ví dụ output:
{
"status": "success",
"result": {...},
"execution_time_ms": 245,
"code_executed": "..."
}
"""
Tạo Code Interpreter Agent
code_interpreter = ConversableAgent(
name="Code_Interpreter_Agent",
system_message=CODE_INTERPRETER_SYSTEM_PROMPT,
llm_config={
"config_list": [{
"model": code_interpreter_config["model"],
"api_key": os.environ["HOLYSHEEP_API_KEY"],
"base_url": os.environ["OPENAI_BASE_URL"],
"price": code_interpreter_config["price"],
"temperature": code_interpreter_config["temperature"],
"max_tokens": code_interpreter_config["max_tokens"],
}],
"timeout": code_interpreter_config["timeout"],
"cache_seed": code_interpreter_config["seed"],
},
max_consecutive_auto_reply=5,
code_execution_config={
"work_dir": "code_workspace",
"use_docker": False, # Set True trong production
"executor": "local", # Hoặc "subprocess" cho isolation
},
human_input_mode="NEVER",
)
print("✓ Code Interpreter Agent khởi tạo thành công!")
print(f" Model: {code_interpreter_config['model']}")
print(f" Price: ${code_interpreter_config['price'][0]}/MTok input, ${code_interpreter_config['price'][1]}/MTok output")Cấu Hình Orchestrator Agent — Điều Phối Multi-Agent
Orchestrator là "bộ não" điều phối, quyết định agent nào xử lý request nào. Đây là phần quan trọng nhất của kiến trúc:
from autogen import UserProxyAgent
User Proxy - nhận input từ user
user_proxy = UserProxyAgent(
name="User_Proxy",
human_input_mode="NEVER",
max_consecutive_reply=10,
code_execution_config={
"work_dir": "code_workspace",
"use_docker": False,
},
)
Orchestrator Agent - điều phối các task
ORCHESTRATOR_SYSTEM_PROMPT = """Bạn là Orchestrator Agent của hệ thống AutoGen.
Nhiệm vụ của bạn là:
1. Phân tích yêu cầu của user
2. Quyết định cần gọi agent nào
3. Tổng hợp kết quả từ nhiều agent
4. Đảm bảo output cuối cùng rõ ràng, hữu ích
Các agent có sẵn:
- Code_Interpreter_Agent: Viết và chạy code Python
- Data_Retriever_Agent: Truy vấn dữ liệu từ database
Luôn reply với JSON format sau:
{
"intent": "phân loại intent (code_execution, data_query, general)",
"agent_to_call": "tên agent cần gọi",
"task_description": "mô tả task chi tiết cho agent",
"priority": "high/medium/low"
}"""
orchestrator = ConversableAgent(
name="Orchestrator_Agent",
system_message=ORCHESTRATOR_SYSTEM_PROMPT,
llm_config={
"config_list": [{
"model": "gpt-4.1",
"api_key": os.environ["HOLYSHEEP_API_KEY"],
"base_url": os.environ["OPENAI_BASE_URL"],
"price": [0.008, 0.024], # Giá HolySheep
"temperature": 0.7,
}],
},
human_input_mode="NEVER",
)
Cấu hình Group Chat cho multi-agent
group_chat = GroupChat(
agents=[user_proxy, orchestrator, code_interpreter],
messages=[],
max_round=15,
speaker_selection_method="auto",
)
group_chat_manager = GroupChatManager(
groupchat=group_chat,
llm_config={
"config_list": [{
"model": "gpt-4.1",
"api_key": os.environ["HOLYSHEEP_API_KEY"],
"base_url": os.environ["OPENAI_BASE_URL"],
}],
},
)
print("✓ Multi-Agent System khởi tạo thành công!")Thực Thi Task — Ví Dụ Phân Tích Dữ Liệu
Bây giờ hãy test hệ thống với một task thực tế. Tôi sẽ demo phân tích dữ liệu bán hàng — use case phổ biến nhất mà tôi gặp:
import time
import asyncio
async def run_code_interpreter_task(user_query: str):
"""Thực thi task thông qua AutoGen multi-agent system"""
start_time = time.time()
# Khởi tạo chat
chat_result = await code_interpreter.a_initiate_chat(
recipient=user_proxy,
message=user_query,
max_turns=5,
summary_method="last_msg",
)
execution_time = (time.time() - start_time) * 1000 # Convert to ms
return {
"chat_result": chat_result,
"execution_time_ms": round(execution_time, 2),
"cost_estimate": calculate_cost(chat_result, code_interpreter_config),
}
def calculate_cost(chat_result, config):
"""Ước tính chi phí dựa trên token usage"""
# Lấy token usage từ chat history
try:
total_input_tokens = 0
total_output_tokens = 0
for msg in chat_result.chat_history:
if "usage" in msg:
total_input_tokens += msg["usage"].get("prompt_tokens", 0)
total_output_tokens += msg["usage"].get("completion_tokens", 0)
input_cost = (total_input_tokens / 1_000_000) * config["price"][0]
output_cost = (total_output_tokens / 1_000_000) * config["price"][1]
return {
"input_tokens": total_input_tokens,
"output_tokens": total_output_tokens,
"total_cost_usd": round(input_cost + output_cost, 6),
}
except:
return {"error": "Không thể tính chi phí"}
Test với query phân tích dữ liệu
test_queries = [
"Tạo một DataFrame với 100 dòng dữ liệu sales, bao gồm columns: date, product, quantity, price. Sau đó tính tổng doanh thu và vẽ biểu đồ.",
"Viết code tính Fibonacci số thứ 50, so sánh 3 phương pháp: recursive, iterative, matrix exponentiation.",
]
async def main():
for i, query in enumerate(test_queries):
print(f"\n{'='*60}")
print(f"TEST {i+1}: {query[:50]}...")
print('='*60)
result = await run_code_interpreter_task(query)
print(f"\n✓ Hoàn thành trong {result['execution_time_ms']}ms")
print(f" Chi phí ước tính: ${result['cost_estimate'].get('total_cost_usd', 'N/A')}")
print(f" Input tokens: {result['cost_estimate'].get('input_tokens', 'N/A')}")
print(f" Output tokens: {result['cost_estimate'].get('output_tokens', 'N/A')}")
Chạy async
asyncio.run(main())Streaming Response — Giảm Perceived Latency
Trong production, streaming response là MUST-HAVE để user không cảm thấy "chờ đợi". Code bên dưới implement streaming với HolySheep:
import threading
import queue
class StreamingCodeInterpreter:
"""Code Interpreter với streaming response - cải thiện UX đáng kể"""
def __init__(self, api_key: str, model: str = "gpt-4.1"):
self.api_key = api_key
self.model = model
self.base_url = "https://api.holysheep.ai/v1"
self.message_queue = queue.Queue()
def stream_execute(self, code: str, callback=None):
"""Execute code với streaming output"""
def stream_worker():
"""Worker thread cho streaming"""
full_response = ""
# Sử dụng HolySheep API với streaming
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
}
payload = {
"model": self.model,
"messages": [
{"role": "system", "content": "Execute this Python code and explain: "},
{"role": "user", "content": code}
],
"stream": True,
"temperature": 0.3,
"max_tokens": 4096,
}
# Call API với streaming (sử dụng requests hoặc httpx)
import httpx
with httpx.Client(timeout=120.0) as client:
with client.stream("POST", f"{self.base_url}/chat/completions",
headers=headers, json=payload) as response:
for chunk in response.iter_lines():
if chunk:
# Parse SSE chunk
if chunk.startswith("data: "):
data = chunk[6:]
if data == "[DONE]":
break
# Parse delta content
# ... (parse JSON and extract content)
delta = "..." # Extracted content
full_response += delta
if callback:
callback(delta)
self.message_queue.put(full_response)
# Chạy worker thread
thread = threading.Thread(target=stream_worker)
thread.start()
return self.message_queue
Usage example
interpreter = StreamingCodeInterpreter(api_key="YOUR_HOLYSHEEP_API_KEY")
def print_chunk(chunk):
print(chunk, end="", flush=True)
result_queue = interpreter.stream_execute(
"Calculate the sum of 1 to 1000 using Python",
callback=print_chunk
)
final_result = result_queue.get()
print(f"\n\nFinal result queued: {final_result[:100]}...")Error Handling và Retry Logic
Một hệ thống production cần xử lý lỗi chủ động. Đây là pattern mà tôi áp dụng cho tất cả các deployment:
import time
from typing import Optional, Callable
from dataclasses import dataclass
from enum import Enum
class ErrorType(Enum):
RATE_LIMIT = "rate_limit"
TIMEOUT = "timeout"
API_ERROR = "api_error"
CODE_ERROR = "code_error"
INVALID_RESPONSE = "invalid_response"
@dataclass
class ExecutionResult:
success: bool
result: Optional[str] = None
error: Optional[str] = None
error_type: Optional[ErrorType] = None
retry_count: int = 0
latency_ms: float = 0.0
class ResilientCodeInterpreter:
"""Code Interpreter với retry logic và error handling"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.max_retries = 3
self.retry_delays = [1, 5, 15] # Exponential backoff (seconds)
def execute_with_retry(self, code: str, context: dict = None) -> ExecutionResult:
"""Execute code với automatic retry"""
last_error = None
start_time = time.time()
for attempt in range(self.max_retries):
try:
result = self._execute_single_attempt(code, context)
latency = (time.time() - start_time) * 1000
return ExecutionResult(
success=True,
result=result,
retry_count=attempt,
latency_ms=round(latency, 2),
)
except Exception as e:
last_error = str(e)
error_type = self._classify_error(e)
# Nếu là rate limit, chờ và retry
if error_type == ErrorType.RATE_LIMIT and attempt < self.max_retries - 1:
delay = self.retry_delays[attempt]
print(f"Rate limit hit. Retrying in {delay}s (attempt {attempt + 1}/{self.max_retries})")
time.sleep(delay)
continue
# Các lỗi khác không retry
break
# Tất cả retries failed
return ExecutionResult(
success=False,
error=last_error,
error_type=error_type,
retry_count=self.max_retries,
latency_ms=(time.time() - start_time) * 1000,
)
def _execute_single_attempt(self, code: str, context: dict = None) -> str:
"""Single execution attempt - implementation here"""
# Implementation using httpx or openai library
# Return the execution result
pass
def _classify_error(self, error: Exception) -> ErrorType:
"""Phân loại loại lỗi để xử lý phù hợp"""
error_str = str(error).lower()
if "429" in error_str or "rate limit" in error_str:
return ErrorType.RATE_LIMIT
elif "timeout" in error_str or "timed out" in error_str:
return ErrorType.TIMEOUT
elif "500" in error_str or "502" in error_str or "503" in error_str:
return ErrorType.API_ERROR
elif "syntax" in error_str or "indentation" in error_str:
return ErrorType.CODE_ERROR
else:
return ErrorType.INVALID_RESPONSE
Usage với monitoring
interpreter = ResilientCodeInterpreter(api_key="YOUR_HOLYSHEEP_API_KEY")
result = interpreter.execute_with_retry("""
import pandas as pd
import numpy as np
Tạo sample data
df = pd.DataFrame({
'date': pd.date_range('2024-01-01', periods=100),
'value': np.random.randn(100).cumsum()
})
Phân tích cơ bản
print(f"Mean: {df['value'].mean():.2f}")
print(f"Std: {df['value'].std():.2f}")
print(f"Min: {df['value'].min():.2f}")
print(f"Max: {df['value'].max():.2f}")
""")
if result.success:
print(f"✓ Thành công sau {result.retry_count} retries")
print(f" Latency: {result.latency_ms}ms")
print(f" Result: {result.result}")
else:
print(f"✗ Thất bại: {result.error}")
print(f" Error Type: {result.error_type.value}")
print(f" Retries: {result.retry_count}")Monitoring và Logging — Production Essential
Để vận hành ổn định, bạn cần monitoring chi tiết. Tôi sử dụng structured logging với metrics:
import json
import logging
from datetime import datetime
from typing import Dict, Any
from dataclasses import dataclass, asdict
@dataclass
class ExecutionMetrics:
"""Metrics cho mỗi execution"""
timestamp: str
request_id: str
model: str
input_tokens: int
output_tokens: int
latency_ms: float
cost_usd: float
success: bool
error_type: str = None
error_message: str = None
def to_log_line(self) -> str:
"""Convert sang JSON log line cho ELK/Splunk"""
return json.dumps(asdict(self))
class MetricsLogger:
"""Logger cho AutoGen execution metrics"""
def __init__(self, log_file: str = "autogen_metrics.log"):
self.log_file = log_file
self.logger = logging.getLogger("AutoGenMetrics")
self.logger.setLevel(logging.INFO)
# File handler
fh = logging.FileHandler(log_file)
fh.setFormatter(logging.Formatter('%(message)s'))
self.logger.addHandler(fh)
# Console handler
ch = logging.StreamHandler()
ch.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))
self.logger.addHandler(ch)
def log_execution(self, metrics: ExecutionMetrics):
"""Log execution metrics"""
# Log to file (structured)
self.logger.info(metrics.to_log_line())
# Console output (human readable)
status = "✓" if metrics.success else "✗"
print(f"{status} [{metrics.request_id}] {metrics.model} | "
f"Tokens: {metrics.input_tokens}/{metrics.output_tokens} | "
f"Latency: {metrics.latency_ms}ms | "
f"Cost: ${metrics.cost_usd:.6f}")
def aggregate_daily_stats(self) -> Dict[str, Any]:
"""Tổng hợp stats hàng ngày từ log file"""
total_requests = 0
total_cost = 0.0
total_latency = 0.0
success_count = 0
with open(self.log_file, 'r') as f:
for line in f:
try:
data = json.loads(line.strip())
total_requests += 1
total_cost += data.get('cost_usd', 0)
total_latency += data.get('latency_ms', 0)
if data.get('success'):
success_count += 1
except:
continue
return {
"total_requests": total_requests,
"total_cost_usd": round(total_cost, 6),
"avg_latency_ms": round(total_latency / max(total_requests, 1), 2),
"success_rate": round(success_count / max(total_requests, 1) * 100, 2),
}
Sử dụng metrics logger
metrics_logger = MetricsLogger()
Log một execution
metrics = ExecutionMetrics(
timestamp=datetime.now().isoformat(),
request_id="req_123456",
model="gpt-4.1",
input_tokens=500,
output_tokens=1200,
latency_ms=850.5,
cost_usd=0.0248, # 0.5K * $0.008 + 1.2K * $0.016
success=True,
)
metrics_logger.log_execution(metrics)
Tổng hợp stats
stats = metrics_logger.aggregate_daily_stats()
print(f"\n📊 Daily Stats: {stats['total_requests']} requests, "
f"${stats['total_cost_usd']} total cost, "
f"{stats['success_rate']}% success rate")So Sánh Chi Phí — HolySheep vs OpenAI
Đây là lý do quan trọng nhất tôi chuyển sang HolySheep cho các dự án production:
| Model | Provider | Giá Input ($/MTok) | Giá Output ($/MTok) | Tiết kiệm |
|---|---|---|---|---|
| GPT-4.1 | OpenAI | $2.50 | $10.00 | - |
| GPT-4.1 | HolySheep | $0.008 | $0.024 | 96.7% |
| Claude Sonnet 4.5 | Anthropic | $3.00 | $15.00 | - |
| Claude Sonnet 4.5 | HolySheep | $0.015 | $0.075 | 95% |
| DeepSeek V3.2 | HolySheep | $0.00042 | $0.00168 | 99.9% |
Với use case phân tích dữ liệu tự động (khoảng 10,000 requests/ngày, trung bình 1000 tokens/request), tiết kiệm hơn $8,000/tháng khi dùng HolySheep thay vì OpenAI.
Lỗi Thường Gặp và Cách Khắc Phục
Qua kinh nghiệm triển khai thực tế, đây là những lỗi phổ biến nhất và giải pháp đã được kiểm chứng:
1. Lỗi "Authentication Error" hoặc "Invalid API Key"
# ❌ SAI - Key bị strip khoảng trắng thừa
os.environ["OPENAI_API_KEY"] = " YOUR_API_KEY " # Có space!
✓ ĐÚNG - Strip whitespace và validate format
def validate_and_set_api_key(key: str) -> bool:
key = key.strip()
# HolySheep key format: sk-xxxx... hoặc holy_xxxx...
if not key:
raise ValueError("API key không được để trống")
if len(key) < 20:
raise ValueError("API key quá ngắn, vui lòng kiểm tra lại")
if not (key.startswith("sk-") or key.startswith("holy_")):
raise ValueError("API key format không hợp lệ. Key phải bắt đầu bằng 'sk-' hoặc 'holy_'")
os.environ["OPENAI_API_KEY"] = key
return True
Sử dụng
try:
validate_and_set_api_key("YOUR_HOLYSHEEP_API_KEY")
print("✓ API Key validated successfully")
except ValueError as e:
print(f"✗ Error: {e}")2. Lỗi Rate Limit (429 Error) - Retry không hoạt động
# ❌ SAI - Retry ngay lập tức, không có backoff
for _ in range(3):
response = call_api()
if response.status_code != 429:
break
time.sleep(1) # Không đủ delay!
✓ ĐÚNG - Exponential backoff với jitter
import random
import time
import httpx
def call_api_with_retry(url: str, payload: dict, api_key: str, max_retries: int = 5):
"""API call với exponential backoff"""
for attempt in range(max_retries):
try:
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
}
with httpx.Client(timeout=120.0) as client:
response = client.post(url, json=payload, headers=headers)
if response.status_code == 200:
return response.json()
elif response.status_code == 429:
# Rate limit - lấy retry-after từ header hoặc tính toán
retry_after = response.headers.get("Retry-After")
if retry_after:
wait_time = int(retry_after)
else:
# Exponential backoff: 1, 2, 4, 8, 16 giây
wait_time = 2 ** attempt
# Thêm jitter (±20%) để tránh thundering herd
jitter = wait_time * 0.2 * (2 * random.random() - 1)
actual_wait = wait_time + jitter
print(f"Rate limit hit. Waiting {actual_wait:.1f}s before retry...")
time.sleep(actual_wait)
elif response.status_code >= 500:
# Server error - retry
wait_time = 2 ** attempt
print(f"Server error {response.status_code}. Retrying in {wait_time}s...")
time.sleep(wait_time)
else:
# Client error - không retry
raise Exception(f"API Error {response.status_code}: {response.text}")
except httpx.TimeoutException:
print(f"Timeout on attempt {attempt + 1}. Retrying...")
time.sleep(2 ** attempt)
raise Exception(f"Failed after {max_retries} retries")3. Lỗi Code Execution Timeout - Code chạy quá lâu
# ❌ SAI - Không có timeout cho code execution
result = code_interpreter.generate_response(user_input)
✓ ĐÚNG - Timeout với graceful handling
import signal
from contextlib import contextmanager
class CodeExecutionTimeout(Exception):
"""Exception khi code execution vượt quá timeout"""
pass
@contextlib.contextmanager
def timeout_context(seconds: int):
"""Context manager cho timeout"""
def signal_handler(signum, frame):
raise CodeExecutionTimeout(f"Code execution exceeded {seconds}s timeout")
# Set the signal handler
old_handler = signal.signal(signal.SIGALRM, signal_handler)
signal.alarm(seconds)
try:
yield
finally:
signal.alarm(0)
signal.signal(signal.SIGALRM, old_handler)
def execute_code_with_timeout(code: str, timeout_seconds: int = 30) -> dict:
"""Execute Python code với timeout protection"""
try:
with timeout_context(timeout_seconds):
# Thực thi code trong sandbox
result = execute_in_sandbox(code)
return {
"success": True,
"result": result,
"execution_time_ms": result.get("execution_time", 0),
"timeout": False,
}
except CodeExecutionTimeout as e:
return {
"success": False,
"error": str(e),
"timeout": True,
"timeout_seconds": timeout_seconds,
"suggestion": "Code quá phức tạp. Thử chia nhỏ thành các bước hoặc tối ưu thuật toán.",
}
except Exception as e:
return {
"success": False,
"error": str(e),
"timeout": False,
}
Test với code chạy lâu
test_code = """
Code phức tạp - sẽ timeout nếu không tối ưu
result = sum(i**2 for i in range(10**7))
print(f"Result: {result}")
"""
result = execute_code_with_timeout(test_code, timeout_seconds=10)
if result["timeout"]:
print(f"⚠️ Code execution timeout after {result['timeout_seconds']}s")
print(f"Suggestion: {result['suggestion']}")4. Lỗi Invalid Response Format - Model trả về không parse được
# ❌ SAI - Không validate response format
response = model.generate(prompt)
data = json.loads(response) # Có thể fail!
✓ ĐÚNG - Validate và sanitize response
import re