تست و دیباگ Agent — چطور بفهمی درست کار می‌کنه

مقدمه: «فکر کنم کار می‌کنه» کافی نیست

با نرم‌افزار معمولی، تست ساده‌ست: ورودی مشخص بده، خروجی مشخص بگیر. اگه ۲+۲ شد ۴، درسته. اگه نشد، غلطه.

با Agent ها قضیه فرق داره. خروجی Agent غیرقطعیه (non-deterministic) — همون ورودی رو بدی، ممکنه دو بار جواب متفاوت بگیری. چطور تست کنی چیزی رو که هر بار جواب متفاوت می‌ده؟

تو این اپیزود تکنیک‌های تست و دیباگ Agent ها رو یاد می‌گیری — از لاگ گرفته تا ابزارهای حرفه‌ای.

چرا تست Agent سخته؟

۱. خروجی غیرقطعی

LLM هر بار ممکنه متن متفاوتی تولید کنه. نمی‌تونی assertEqual بزنی و بگی «خروجی باید دقیقاً این باشه».

۲. زنجیره تصمیمات

Agent یه پاسخ ساده نمی‌ده — یه سری تصمیم می‌گیره: کدوم ابزار رو استفاده کنم؟ چه پارامتری بدم؟ نتیجه رو چطور تفسیر کنم؟ هر کدوم ممکنه اشتباه بشه.

۳. وابستگی به سرویس خارجی

Agent به LLM API، ابزارهای خارجی و دیتابیس وابسته‌ست. هر کدوم ممکنه تغییر کنه یا از دسترس خارج بشه.

سطح ۱: لاگ همه چیز (Trace Logging)

اول از همه، باید ببینی Agent داره چیکار می‌کنه. بدون لاگ، دیباگ Agent مثل تعمیر ماشین تو تاریکیه.

import json
import time
from datetime import datetime
from functools import wraps

class AgentTracer:
    """همه اقدامات Agent رو ثبت می‌کنه."""
    
    def __init__(self):
        self.traces = []
        self.current_trace = None
    
    def start_trace(self, user_input: str):
        """شروع یه trace جدید."""
        self.current_trace = {
            "id": f"trace_{int(time.time())}",
            "timestamp": datetime.now().isoformat(),
            "input": user_input,
            "steps": [],
            "total_tokens": 0,
            "total_cost": 0.0,
            "total_time": 0.0,
        }
        self._start_time = time.time()
    
    def log_step(self, step_type: str, data: dict):
        """یه مرحله ثبت می‌کنه."""
        step = {
            "type": step_type,
            "timestamp": datetime.now().isoformat(),
            "data": data,
        }
        self.current_trace["steps"].append(step)
        
        # ثبت توکن و هزینه
        if "usage" in data:
            tokens = data["usage"].get("total_tokens", 0)
            self.current_trace["total_tokens"] += tokens
            # تخمین هزینه (gpt-4o-mini)
            self.current_trace["total_cost"] += (
                tokens * 0.00015 / 1000
            )
    
    def end_trace(self, output: str):
        """پایان trace."""
        self.current_trace["output"] = output
        self.current_trace["total_time"] = (
            time.time() - self._start_time
        )
        self.traces.append(self.current_trace)
        return self.current_trace
    
    def print_trace(self, trace: dict = None):
        """trace رو خوانا چاپ می‌کنه."""
        t = trace or self.current_trace
        print(f"\n{'='*60}")
        print(f"Trace: {t['id']}")
        print(f"Input: {t['input']}")
        print(f"{'='*60}")
        
        for i, step in enumerate(t["steps"], 1):
            print(f"\n--- Step {i}: {step['type']} ---")
            for key, value in step["data"].items():
                if key == "usage":
                    continue
                val_str = str(value)[:200]
                print(f"  {key}: {val_str}")
        
        print(f"\n{'='*60}")
        print(f"Output: {t['output'][:200]}")
        print(f"Tokens: {t['total_tokens']}")
        print(f"Cost: ${t['total_cost']:.4f}")
        print(f"Time: {t['total_time']:.2f}s")
        print(f"{'='*60}\n")

# استفاده تو Agent
tracer = AgentTracer()

async def process_with_tracing(user_input: str) -> str:
    tracer.start_trace(user_input)
    
    # لاگ فراخوانی LLM
    tracer.log_step("llm_call", {
        "model": "gpt-4o-mini",
        "messages_count": 5,
        "tools_available": ["weather", "calculate"],
    })
    
    # لاگ استفاده از ابزار
    tracer.log_step("tool_call", {
        "tool": "get_weather",
        "args": {"city": "Tehran"},
        "result": "آفتابی، ۲۸ درجه",
    })
    
    # لاگ پاسخ نهایی
    output = "هوای تهران آفتابیه و ۲۸ درجه‌ست."
    trace = tracer.end_trace(output)
    tracer.print_trace(trace)
    
    return output

سطح ۲: تست ابزارها (Unit Testing Tools)

ابزارها قطعی هستن — ورودی مشخص، خروجی مشخص. اینا رو می‌تونی راحت unit test بنویسی.

import pytest

# تست ابزار ماشین‌حساب
class TestCalculateTool:
    def test_basic_addition(self):
        result = calculate("2 + 3")
        assert "5" in result
    
    def test_complex_expression(self):
        result = calculate("(10 + 5) * 2")
        assert "30" in result
    
    def test_division(self):
        result = calculate("10 / 3")
        assert "3.33" in result
    
    def test_invalid_expression(self):
        result = calculate("hello")
        assert "خطا" in result
    
    def test_dangerous_input(self):
        """نباید اجازه اجرای کد بده."""
        result = calculate("__import__('os').system('ls')")
        assert "خطا" in result or "مجاز" in result


# تست ابزار آب‌وهوا
class TestWeatherTool:
    @pytest.mark.asyncio
    async def test_valid_city(self):
        result = await get_weather("Tehran")
        assert "دما" in result
        assert "°C" in result
    
    @pytest.mark.asyncio
    async def test_invalid_city(self):
        result = await get_weather("XYZNotACity")
        assert "خطا" in result


# تست ابزار زمان
class TestTimeTool:
    def test_tehran_time(self):
        result = get_current_time("Asia/Tehran")
        assert len(result) > 0
        assert "خطا" not in result
    
    def test_invalid_timezone(self):
        result = get_current_time("Invalid/Zone")
        assert "نامعتبر" in result

سطح ۳: تست رفتار Agent (Behavioral Testing)

اینجا جالب می‌شه. نمی‌تونیم خروجی دقیق رو چک کنیم، ولی می‌تونیم رفتار رو چک کنیم.

import pytest
from agent import process_message

class TestAgentBehavior:
    """تست رفتار Agent — نه خروجی دقیق."""
    
    @pytest.mark.asyncio
    async def test_uses_weather_tool_for_weather_query(self):
        """وقتی هوا رو می‌پرسی، باید ابزار آب‌وهوا استفاده کنه."""
        tracer.start_trace("هوای تهران چطوره؟")
        result = await process_message(
            chat_id=999, 
            user_message="هوای تهران چطوره؟",
            user_name="Test"
        )
        trace = tracer.end_trace(result)
        
        # چک کن ابزار آب‌وهوا صدا زده شده
        tool_steps = [
            s for s in trace["steps"] 
            if s["type"] == "tool_call"
        ]
        assert len(tool_steps) > 0
        assert tool_steps[0]["data"]["tool"] == "get_weather"
    
    @pytest.mark.asyncio
    async def test_responds_in_persian(self):
        """جواب باید فارسی باشه."""
        result = await process_message(
            chat_id=999,
            user_message="سلام، حالت چطوره؟",
            user_name="Test"
        )
        
        # بررسی وجود کاراکترهای فارسی
        persian_chars = set("ابپتثجچحخدذرزژسشصضطظعغفقکگلمنوهی")
        has_persian = any(c in persian_chars for c in result)
        assert has_persian, f"پاسخ فارسی نیست: {result}"
    
    @pytest.mark.asyncio
    async def test_says_dont_know_for_unknown(self):
        """برای سوالات نامربوط باید بگه نمی‌دونم."""
        result = await process_message(
            chat_id=999,
            user_message="شماره تلفن خونه مادربزرگ من چنده؟",
            user_name="Test"
        )
        
        negative_phrases = ["نمی‌دونم", "نمی‌تونم", "اطلاعی ندارم"]
        has_negative = any(p in result for p in negative_phrases)
        assert has_negative, f"باید بگه نمی‌دونه: {result}"
    
    @pytest.mark.asyncio
    async def test_remembers_context(self):
        """باید مکالمه قبلی رو یادش باشه."""
        chat_id = 888
        
        # پیام اول
        await process_message(
            chat_id=chat_id,
            user_message="اسم من علیه",
            user_name="Test"
        )
        
        # پیام دوم
        result = await process_message(
            chat_id=chat_id,
            user_message="اسم من چیه؟",
            user_name="Test"
        )
        
        assert "علی" in result, f"حافظه کار نکرد: {result}"

سطح ۴: Golden Dataset (مجموعه طلایی)

یه مجموعه از سوالات و جواب‌های مورد انتظار درست کن. هر بار که Agent رو تغییر می‌دی، روی این مجموعه تستش کن.

import json

GOLDEN_DATASET = [
    {
        "input": "هوای اصفهان چطوره؟",
        "expected_tool": "get_weather",
        "expected_contains": ["اصفهان", "درجه"],
        "category": "weather",
    },
    {
        "input": "۲۵ ضرب‌در ۴ چنده؟",
        "expected_tool": "calculate",
        "expected_contains": ["100"],
        "category": "math",
    },
    {
        "input": "سلام خوبی؟",
        "expected_tool": None,
        "expected_contains": [],
        "category": "greeting",
    },
    {
        "input": "ساعت چنده الان؟",
        "expected_tool": "get_current_time",
        "expected_contains": [":"],
        "category": "time",
    },
]


async def run_golden_tests():
    """تست‌های طلایی رو اجرا می‌کنه."""
    results = []
    
    for test_case in GOLDEN_DATASET:
        tracer.start_trace(test_case["input"])
        result = await process_message(
            chat_id=777,
            user_message=test_case["input"],
            user_name="GoldenTest"
        )
        trace = tracer.end_trace(result)
        
        # بررسی ابزار
        tool_used = None
        tool_steps = [
            s for s in trace["steps"] 
            if s["type"] == "tool_call"
        ]
        if tool_steps:
            tool_used = tool_steps[0]["data"]["tool"]
        
        tool_correct = tool_used == test_case["expected_tool"]
        
        # بررسی محتوا
        content_correct = all(
            keyword in result 
            for keyword in test_case["expected_contains"]
        )
        
        passed = tool_correct and content_correct
        
        results.append({
            "input": test_case["input"],
            "category": test_case["category"],
            "passed": passed,
            "tool_expected": test_case["expected_tool"],
            "tool_actual": tool_used,
            "output_preview": result[:100],
        })
        
        status = "PASS" if passed else "FAIL"
        print(f"[{status}] {test_case['input'][:40]}")
    
    # خلاصه
    total = len(results)
    passed = sum(1 for r in results if r["passed"])
    print(f"\n{passed}/{total} tests passed "
          f"({passed/total*100:.0f}%)")
    
    return results

LangSmith — ابزار حرفه‌ای مانیتورینگ

LangSmith از تیم LangChain هست ولی با هر فریمورکی کار می‌کنه. یه داشبورد وب بهت می‌ده که همه فراخوانی‌های LLM رو ببینی، دیباگ کنی و ارزیابی کنی.

# نصب: pip install langsmith
# تنظیم environment variables:
# LANGCHAIN_API_KEY=your_key
# LANGCHAIN_TRACING_V2=true
# LANGCHAIN_PROJECT=my-agent

from langsmith import traceable
from openai import OpenAI

client = OpenAI()

@traceable(name="agent_response")
def get_response(user_input: str) -> str:
    """هر فراخوانی خودکار تو LangSmith ثبت می‌شه."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "دستیار فارسی"},
            {"role": "user", "content": user_input}
        ]
    )
    return response.choices[0].message.content

# هر بار get_response صدا بزنی، 
# trace کامل تو داشبورد LangSmith ظاهر می‌شه:
# - ورودی و خروجی
# - مصرف توکن
# - زمان اجرا
# - خطاها

چرا LangSmith مفیده؟

• همه trace ها رو تو یه داشبورد می‌بینی
• می‌تونی فیلتر کنی — مثلاً فقط trace های با خطا
• هزینه هر مکالمه رو نشون می‌ده
• می‌تونی dataset بسازی و ارزیابی خودکار بذاری

Regression Testing — مطمئن شو چیزی نشکسته

class RegressionTest:
    """تست رگرسیون — مطمئن شو آپدیت جدید 
    چیزی رو خراب نکرده."""
    
    def __init__(self, baseline_file: str):
        self.baseline_file = baseline_file
    
    def save_baseline(self, results: list):
        """نتایج فعلی رو به عنوان baseline ذخیره کن."""
        with open(self.baseline_file, "w") as f:
            json.dump(results, f, ensure_ascii=False)
    
    def compare(self, new_results: list) -> dict:
        """نتایج جدید رو با baseline مقایسه کن."""
        with open(self.baseline_file) as f:
            baseline = json.load(f)
        
        regressions = []
        improvements = []
        
        for old, new in zip(baseline, new_results):
            if old["passed"] and not new["passed"]:
                regressions.append({
                    "input": new["input"],
                    "was": "PASS",
                    "now": "FAIL",
                })
            elif not old["passed"] and new["passed"]:
                improvements.append({
                    "input": new["input"],
                    "was": "FAIL",
                    "now": "PASS",
                })
        
        return {
            "regressions": regressions,
            "improvements": improvements,
            "regression_count": len(regressions),
            "improvement_count": len(improvements),
        }

# استفاده:
# ۱. اول یه بار baseline بساز
# results = await run_golden_tests()
# reg_test = RegressionTest("baseline.json")
# reg_test.save_baseline(results)

# ۲. بعد از هر تغییر، مقایسه کن
# new_results = await run_golden_tests()
# comparison = reg_test.compare(new_results)
# if comparison["regression_count"] > 0:
#     print("هشدار! رگرسیون پیدا شد!")

نکات عملی دیباگ

۱. همیشه verbose رو روشن کن: موقع توسعه، همه لاگ‌ها رو ببین. هر فراخوانی LLM، هر ابزار، هر تصمیم.

۲. مشکل رو ایزوله کن: اگه Agent اشتباه عمل می‌کنه، اول ببین مشکل از LLM ه یا ابزار. ابزار رو جدا تست کن. اگه ابزار درسته، مشکل از prompt یا مدله.

۳. Temperature رو کم کن: موقع تست، temperature رو ۰ بذار تا خروجی تا حد ممکن قابل تکرار باشه.

۴. مکالمات واقعی رو ذخیره کن: مکالمات کاربران واقعی رو (با رعایت حریم خصوصی) ذخیره کن و ازشون برای بهبود Agent استفاده کن.

۵. A/B تست بزن: دو نسخه Agent رو همزمان اجرا کن و ببین کدوم بهتره.

جمع‌بندی

• تست Agent سخته ولی ضروریه
• لاگ همه چیز — بدون لاگ دیباگ غیرممکنه
• ابزارها رو unit test کن — اینا قطعی هستن
• رفتار Agent رو تست کن، نه خروجی دقیق
• Golden dataset بساز و regression test بزن
• LangSmith یا ابزار مشابه برای مانیتورینگ استفاده کن

اپیزود بعدی — اپیزود آخر سری — یه Research Agent کامل از صفر می‌سازیم!