H.Ozgan
a.k.a netologist

LLM & Agent Production Mühendisliği — Principal Seviye Rehber

22 Mart 2026 · netologist · 32 dakika, 6768 kelime · #agents #ai #evals #llm #note #observability #production #python #reliability

Notebook’ta “çalışıyor” olması ile production’da çalışması arasındaki fark tam olarak bu dokümanın konusu: LLM engineering, agent reliability, evaluation ve observability. Geri kalan her şey (framework’ler, prompt kütüphaneleri, wrapper SDK’lar) bu dört sütunun üzerine giydirilmiş süsten ibaret.

İçindekiler

  1. LLM Engineering
    • 1.1 Prompt Design
    • 1.2 Structured Generation
    • 1.3 Tool Calling
    • 1.4 Context Engineering
    • 1.5 Model Selection
    • 1.6 Token Economics
    • 1.7 Streaming
    • 1.8 Fallbacks
  2. Agent Reliability
    • 2.1 Timeouts
    • 2.2 Retries
    • 2.3 Idempotency
    • 2.4 State Persistence
    • 2.5 Failure Recovery
    • 2.6 Guardrails
    • 2.7 Validation
    • 2.8 Human Escalation
  3. Evaluation
  4. Observability
  5. Hepsini Bir Araya Getirmek — Referans Mimari

1. LLM Engineering

1.1 Prompt Design

Prensip: Prompt bir düz yazı değil, olasılıksal bir fonksiyonla yapılmış bir API kontratıdır. Onu kod gibi ele al: versiyonlanmış, test edilmiş, review edilmiş.

Best practices

Design pattern: Prompt Template Registry

from dataclasses import dataclass
from string import Template
from pathlib import Path
import hashlib

@dataclass(frozen=True)
class PromptVersion:
    name: str
    version: str
    template: str
    checksum: str

class PromptRegistry:
    """Tüm promptlar için merkezi, versiyonlanmış tek doğru kaynak.
    Promptları asla business logic içinde inline formatlama."""

    def __init__(self, prompt_dir: Path):
        self._prompts: dict[str, PromptVersion] = {}
        for file in prompt_dir.glob("*.tmpl"):
            name, version = file.stem.rsplit("__", 1)
            text = file.read_text()
            checksum = hashlib.sha256(text.encode()).hexdigest()[:8]
            self._prompts[f"{name}:{version}"] = PromptVersion(name, version, text, checksum)

    def render(self, name: str, version: str, **kwargs) -> str:
        key = f"{name}:{version}"
        pv = self._prompts[key]
        return Template(pv.template).substitute(**kwargs)

# Kullanım
registry = PromptRegistry(Path("prompts/"))
prompt = registry.render("support_triage", "v3", ticket_body="...", policy="...")

Anti-pattern’ler

1.2 Structured Generation

Prensip: Eğer downstream kod çıktıyı parse ediyorsa, model sadece parse edilebilir çıktı üretecek şekilde kısıtlanmalıdır — asla “lütfen valid JSON ile cevap ver” ifadesine tek başına güvenme.

Best practices

Design pattern: Schema-first generation with self-repair

from pydantic import BaseModel, ValidationError, Field
from typing import Literal
import json

class TriageResult(BaseModel):
    category: Literal["billing", "technical", "abuse", "other"]
    severity: Literal["low", "medium", "high", "critical"]
    summary: str = Field(max_length=280)
    requires_human: bool

def generate_structured(client, prompt: str, schema: type[BaseModel], max_repairs: int = 2):
    """Üret + doğrula + kendini onar döngüsü.
    Bozuk bir payload'ın sessizce downstream'e sızmasına asla izin verme."""
    messages = [{"role": "user", "content": prompt}]

    for attempt in range(max_repairs + 1):
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            tools=[{
                "name": "emit_result",
                "input_schema": schema.model_json_schema(),
            }],
            tool_choice={"type": "tool", "name": "emit_result"},
            messages=messages,
        )
        tool_call = next(b for b in response.content if b.type == "tool_use")
        try:
            return schema.model_validate(tool_call.input)
        except ValidationError as e:
            if attempt == max_repairs:
                raise
            # Validasyon hatasını geri besle — sadece kör kör retry değil, kendini onarım
            messages.append({"role": "assistant", "content": response.content})
            messages.append({
                "role": "user",
                "content": f"Çıktın validasyonu geçemedi: {e}. Düzelt ve emit_result'ı tekrar çağır."
            })

Anti-pattern’ler

1.3 Tool Calling

Prensip: Bir tool tanımı, üzerinde modelin akıl yürüttüğü bir kontrattır, insanlar için dokümantasyon değil. Spesifikasyondaki her belirsizlik, runtime’da bir hata moduna dönüşür.

Best practices

Design pattern: Büyük toolset’ler için Tool Router

from typing import Protocol

class Tool(Protocol):
    name: str
    description: str
    schema: dict

class ToolRouter:
    """İki aşamalı tool calling: önce ilgili tool *gruplarını* seç,
    sonra yalnızca onları execution çağrısına aç. 100+ tool kayıtlı
    olsa bile working-set'i küçük ve seçim doğruluğunu yüksek tutar."""

    def __init__(self, groups: dict[str, list[Tool]]):
        self.groups = groups

    def route(self, client, user_intent: str) -> list[Tool]:
        group_names = list(self.groups.keys())
        response = client.messages.create(
            model="claude-haiku-4-5",  # routing için ucuz, hızlı model
            max_tokens=50,
            messages=[{
                "role": "user",
                "content": f"Şu talebe göre: '{user_intent}', hangi tool grupları "
                            f"ilgili? Seçenekler: {group_names}. JSON liste olarak cevap ver."
            }],
        )
        selected = json.loads(response.content[0].text)
        tools = []
        for g in selected:
            tools.extend(self.groups.get(g, []))
        return tools

Anti-pattern’ler

1.4 Context Engineering

Prensip: Context window kıt, pahalı ve dikkati sulandıran bir kaynaktır. Daha fazla context, daha fazla doğruluk anlamına gelmez — alakasız context reasoning’i aktif olarak bozar (“lost in the middle” etkisi, gürültüden dikkat dağılması).

Best practices

Design pattern: Context Budgeter

from dataclasses import dataclass, field

@dataclass
class ContextBudget:
    max_tokens: int
    system_reserve: int
    output_reserve: int

    @property
    def available_for_dynamic(self) -> int:
        return self.max_tokens - self.system_reserve - self.output_reserve

class ContextAssembler:
    """Sıkı bir token bütçesi altında nihai prompt context'ini oluşturur,
    relevance skoruna göre önceliklendirir, asla sessizce taşırmaz."""

    def __init__(self, budget: ContextBudget, tokenizer):
        self.budget = budget
        self.tokenizer = tokenizer

    def assemble(self, ranked_chunks: list[tuple[str, float]]) -> str:
        # ranked_chunks: (text, relevance_score), azalan sırada
        selected, used = [], 0
        for text, score in ranked_chunks:
            n = len(self.tokenizer.encode(text))
            if used + n > self.budget.available_for_dynamic:
                break
            selected.append(text)
            used += n
        return "\n---\n".join(selected)

Anti-pattern’ler

1.5 Model Selection

Prensip: Model seçimi, tek bir küresel karar değil, görev başına bir maliyet/latency/kalite optimizasyon problemidir. Principal mühendisler bir kere seçmez, yönlendirir (route eder).

Best practices

Design pattern: Katmanlı Model Router

from enum import Enum

class TaskComplexity(Enum):
    TRIVIAL = "trivial"      # routing, basit classification
    MODERATE = "moderate"    # extraction, tek adımlı reasoning
    COMPLEX = "complex"      # çok adımlı reasoning, planlama, kod üretimi

MODEL_TIER = {
    TaskComplexity.TRIVIAL: "claude-haiku-4-5-20251001",
    TaskComplexity.MODERATE: "claude-sonnet-5",
    TaskComplexity.COMPLEX: "claude-opus-4-8",
}

def select_model(complexity: TaskComplexity, escalate: bool = False) -> str:
    if escalate:
        # başarısızlık sonrası retry'da bir tier yukarı çık, tepede sınırla
        tiers = list(TaskComplexity)
        idx = min(tiers.index(complexity) + 1, len(tiers) - 1)
        complexity = tiers[idx]
    return MODEL_TIER[complexity]

Anti-pattern’ler

1.6 Token Economics

Prensip: Token’lar senin birincil birim maliyetin ve birincil latency sürücündür. Token bütçelerini dashboard’lu ve alarmlı bir cloud altyapı maliyet merkezi gibi ele al.

Best practices

Design pattern: Token Budget Guard

class TokenBudgetExceeded(Exception):
    pass

class SessionBudget:
    """Agent oturumu başına sert tavan — kaçak döngülerin
    bir maliyet olayına dönüşmesini engeller."""

    def __init__(self, max_tokens: int):
        self.max_tokens = max_tokens
        self.used = 0

    def charge(self, input_tokens: int, output_tokens: int):
        self.used += input_tokens + output_tokens
        if self.used > self.max_tokens:
            raise TokenBudgetExceeded(
                f"Oturum {self.used} token kullandı, bütçe {self.max_tokens} idi"
            )

    @property
    def remaining(self) -> int:
        return max(0, self.max_tokens - self.used)

Anti-pattern’ler

1.7 Streaming

Prensip: Streaming sadece bir latency hilesi değil, bir UX ve reliability aracıdır — partial çıktıyı incremental olarak parse edilecek ve doğrulanacak birinci sınıf bir veri yapısı olarak ele al.

Best practices

Design pattern: Incremental validation ile streaming

import json

async def stream_structured_response(client, prompt: str, schema: type[BaseModel]):
    buffer = ""
    async with client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=2048,
        messages=[{"role": "user", "content": prompt}],
    ) as stream:
        async for event in stream:
            if event.type == "content_block_delta":
                buffer += event.delta.text
                yield {"type": "partial", "text": buffer}  # UX: ilerlemeyi göster

        final = await stream.get_final_message()

    try:
        parsed = schema.model_validate_json(buffer)
        yield {"type": "final", "data": parsed}
    except ValidationError as e:
        yield {"type": "error", "detail": str(e)}

Anti-pattern’ler

1.8 Fallbacks

Prensip: Her dış çağrı (model API’si, tool, retrieval backend’i) production’da kesinlikle bir gün başarısız olacaktır. Soru asla “eğer” değil, “sonra ne olacak” ve bu doğaçlama değil, tasarlanmış olmalıdır.

Best practices

Design pattern: Katmanlı Fallback Zinciri

from typing import Callable, Any

class FallbackChain:
    """Her biri kendi circuit breaker'ına sahip sıralı strateji zinciri.
    Başarısızlıkta bir sonraki katmana düşer, çağırana asla ham bir
    exception göstermez."""

    def __init__(self, strategies: list[Callable[..., Any]]):
        self.strategies = strategies

    def execute(self, *args, **kwargs) -> tuple[Any, int]:
        last_error = None
        for tier, strategy in enumerate(self.strategies):
            try:
                return strategy(*args, **kwargs), tier
            except Exception as e:  # noqa: kasıtlı olarak geniş — fallback sınırı
                last_error = e
                continue
        raise RuntimeError(f"Tüm fallback katmanları tükendi: {last_error}")

# Kullanım
chain = FallbackChain([
    lambda q: full_agent_answer(q),        # katman 0: tam kalite
    lambda q: single_call_answer(q),       # katman 1: düşük kalite
    lambda q: cached_template_answer(q),   # katman 2: statik fallback
])
answer, tier_used = chain.execute(user_query)
if tier_used > 0:
    logger.warning("degraded_response", tier=tier_used)

Anti-pattern’ler

2. Agent Reliability

2.1 Timeouts

Prensip: Her network çağrısı, model çağrısı ve tool çağrısının açık bir timeout’a ihtiyacı vardır — “timeout yok” nötr bir durum değil, sınırsız bir yükümlülüktür.

Best practices

Design pattern: Deadline propagation

import time
import asyncio

class Deadline:
    """Her katmanın kendi bağımsız timeout'unu icat etmesi yerine, paylaşılan
    bir wall-clock deadline'ı iç içe geçmiş async çağrılar boyunca iletir."""

    def __init__(self, timeout_s: float):
        self.deadline = time.monotonic() + timeout_s

    @property
    def remaining(self) -> float:
        return max(0.0, self.deadline - time.monotonic())

    def check(self):
        if self.remaining <= 0:
            raise TimeoutError("Deadline aşıldı")

async def run_agent_turn(deadline: Deadline, agent_step):
    deadline.check()
    return await asyncio.wait_for(agent_step(), timeout=deadline.remaining)

async def agent_loop(user_query: str):
    deadline = Deadline(timeout_s=30)
    for step in build_steps(user_query):
        result = await run_agent_turn(deadline, step)
        deadline.check()

Anti-pattern’ler

2.2 Retries

Prensip: Retry’lar geçici hataları başarıya dönüştürür ve kalıcı hataları kademeli yüke dönüştürür. Retry politikası aradaki farkı bilmelidir.

Best practices

Design pattern: Backoff+jitter ile politika güdümlü retry

import random
import time
from dataclasses import dataclass

@dataclass
class RetryPolicy:
    max_attempts: int
    base_delay_s: float
    max_delay_s: float
    retryable: Callable[[Exception], bool]

def with_retry(policy: RetryPolicy):
    def decorator(fn):
        def wrapped(*args, **kwargs):
            last_exc = None
            for attempt in range(policy.max_attempts):
                try:
                    return fn(*args, **kwargs)
                except Exception as e:
                    if not policy.retryable(e) or attempt == policy.max_attempts - 1:
                        raise
                    last_exc = e
                    delay = min(policy.base_delay_s * (2 ** attempt), policy.max_delay_s)
                    delay *= random.uniform(0.5, 1.5)  # jitter
                    logger.warning("retrying", attempt=attempt, delay=delay, error=str(e))
                    time.sleep(delay)
            raise last_exc
        return wrapped
    return decorator

RATE_LIMIT_POLICY = RetryPolicy(
    max_attempts=4, base_delay_s=1.0, max_delay_s=20.0,
    retryable=lambda e: isinstance(e, (RateLimitError, TimeoutError, ServerError)),
)

@with_retry(RATE_LIMIT_POLICY)
def call_llm(prompt): ...

Anti-pattern’ler

2.3 Idempotency

Prensip: Dış durumu değiştiren herhangi bir tool çağrısı, aynı input ile birden fazla kez çalıştırılması güvenli olmalıdır — çünkü retry’lar, timeout’lar ve process çökmeleri kesinlikle yinelenen çağrılara neden olacaktır.

Best practices

Design pattern: Idempotency ledger

import hashlib
from datetime import datetime, timedelta

class IdempotencyLedger:
    """Yan etkili bir eylemi çalıştırmadan önce, bu key altında daha önce
    yapılıp yapılmadığını kontrol et. Production'da Redis/DB tablosu;
    burada gösterim için dict."""

    def __init__(self, store: dict, ttl: timedelta = timedelta(hours=24)):
        self.store = store
        self.ttl = ttl

    def make_key(self, session_id: str, step_id: str, action: str, payload: dict) -> str:
        raw = f"{session_id}:{step_id}:{action}:{sorted(payload.items())}"
        return hashlib.sha256(raw.encode()).hexdigest()

    def execute_once(self, key: str, fn: Callable[[], Any]) -> Any:
        existing = self.store.get(key)
        if existing and existing["expires_at"] > datetime.utcnow():
            return existing["result"]  # zaten yapılmış — cache'lenmiş sonucu döndür, yan etkiyi tekrarlama

        result = fn()
        self.store[key] = {"result": result, "expires_at": datetime.utcnow() + self.ttl}
        return result

# Bir tool çağrısı içinde kullanım
def charge_customer_tool(session_id, step_id, amount):
    key = ledger.make_key(session_id, step_id, "charge_customer", {"amount": amount})
    return ledger.execute_once(key, lambda: payment_api.charge(amount))

Anti-pattern’ler

2.4 State Persistence

Prensip: Bir agent’ın çalışma durumu (konuşma, plan, ara sonuçlar, tool çağrı geçmişi) process yeniden başlatmalarından, deploy’lardan ve çökmelerden hayatta kalmalıdır — bunu in-memory bir kolaylık değil, kalıcı veri olarak ele al.

Best practices

Design pattern: Checkpoint’li state machine

from enum import Enum, auto
from dataclasses import dataclass, field, asdict
import json

class AgentState(Enum):
    PLANNING = auto()
    EXECUTING_TOOL = auto()
    AWAITING_HUMAN = auto()
    DONE = auto()
    FAILED = auto()

@dataclass
class AgentCheckpoint:
    session_id: str
    state: AgentState
    plan: list[str]
    current_step: int
    tool_results: dict = field(default_factory=dict)
    retry_counts: dict = field(default_factory=dict)

class CheckpointStore:
    """Agent durumu için dayanıklı kalıcılık — prod'da Postgres/Redis."""

    def __init__(self, backend):
        self.backend = backend

    def save(self, checkpoint: AgentCheckpoint):
        payload = asdict(checkpoint)
        payload["state"] = checkpoint.state.name
        self.backend.set(checkpoint.session_id, json.dumps(payload))

    def load(self, session_id: str) -> AgentCheckpoint | None:
        raw = self.backend.get(session_id)
        if not raw:
            return None
        data = json.loads(raw)
        data["state"] = AgentState[data["state"]]
        return AgentCheckpoint(**data)

def resume_or_start(store: CheckpointStore, session_id: str, initial_plan):
    checkpoint = store.load(session_id)
    if checkpoint is None:
        checkpoint = AgentCheckpoint(session_id, AgentState.PLANNING, initial_plan, 0)
        store.save(checkpoint)
    return checkpoint

Anti-pattern’ler

2.5 Failure Recovery

Prensip: Failure recovery, try/except: log ve devam et değil, tasarlanmış bir state machine’dir. Her hata modu, açık ve test edilmiş bir kurtarma yoluna ihtiyaç duyar.

Best practices

Design pattern: Recovery-aware executor

class RecoverableExecutionError(Exception):
    def __init__(self, checkpoint: AgentCheckpoint, cause: Exception):
        self.checkpoint = checkpoint
        self.cause = cause

def execute_plan(checkpoint: AgentCheckpoint, store: CheckpointStore):
    try:
        for i in range(checkpoint.current_step, len(checkpoint.plan)):
            step = checkpoint.plan[i]
            result = execute_step(step, checkpoint)  # 2.3'e göre idempotent
            checkpoint.tool_results[step] = result
            checkpoint.current_step = i + 1
            store.save(checkpoint)  # her adımdan sonra checkpoint
        checkpoint.state = AgentState.DONE
        store.save(checkpoint)
        return checkpoint
    except Exception as e:
        checkpoint.state = AgentState.FAILED
        store.save(checkpoint)
        raise RecoverableExecutionError(checkpoint, e)

def recover(session_id: str, store: CheckpointStore):
    checkpoint = store.load(session_id)
    if checkpoint is None:
        raise ValueError("kurtarılacak bir checkpoint yok")
    if checkpoint.state == AgentState.FAILED:
        # checkpoint.current_step'ten devam eder, tamamlanmış adımları tekrarlamaz
        checkpoint.state = AgentState.PLANNING
        return execute_plan(checkpoint, store)

Anti-pattern’ler

2.6 Guardrails

Prensip: Guardrail’ler, agent’ın talimat aldığı şeyden bağımsız olarak yapmasına izin verilen şeyi kısıtlar — prompt injection, model hatası ve kapsam genişlemesine karşı derinlemesine savunma.

Best practices

Design pattern: Katmanlı guardrail pipeline’ı

from dataclasses import dataclass

@dataclass
class GuardrailViolation(Exception):
    layer: str
    reason: str

class GuardrailPipeline:
    """Input, output ve eylem guardrail'lerini bağımsız, birleştirilebilir
    katmanlar olarak uygular. Internal hatada kapalı başarısız olur."""

    def __init__(self, input_checks, output_checks, action_checks):
        self.input_checks = input_checks
        self.output_checks = output_checks
        self.action_checks = action_checks

    def check_input(self, untrusted_content: str):
        for check in self.input_checks:
            try:
                if not check(untrusted_content):
                    raise GuardrailViolation("input", check.__name__)
            except GuardrailViolation:
                raise
            except Exception:
                raise GuardrailViolation("input", f"{check.__name__}_errored")  # kapalı başarısız ol

    def check_action(self, tool_name: str, tool_args: dict, session_scope: dict):
        for check in self.action_checks:
            if not check(tool_name, tool_args, session_scope):
                raise GuardrailViolation("action", f"{tool_name}_denied")

def detect_prompt_injection(text: str) -> bool:
    """İçerik geçerse (güvenliyse) True döner."""
    suspicious_markers = ["ignore previous instructions", "system:", "you are now"]
    return not any(m in text.lower() for m in suspicious_markers)

def within_financial_limit(tool_name, args, scope):
    if tool_name != "issue_refund":
        return True
    return args.get("amount", 0) <= scope.get("max_refund", 0)

Anti-pattern’ler

2.7 Validation

Prensip: Validation, “model token üretti” ile “sistem gerçekler üzerine hareket etti” arasındaki sınırdır. Her sınır geçişi bir validator’a ihtiyaç duyar — input’lar, tool argümanları ve output’lar aynı şekilde.

Best practices

Design pattern: Çok-sınırlı validasyon

from pydantic import BaseModel, field_validator
from datetime import date

class RefundRequest(BaseModel):
    order_id: str
    amount_cents: int
    reason: str

    @field_validator("amount_cents")
    @classmethod
    def sane_amount(cls, v):
        if v <= 0:
            raise ValueError("miktar pozitif olmalı")
        if v > 1_000_000:  # $10.000 — sadece tip geçerliliği değil, business-kural tavanı
            raise ValueError("miktar otomatik onay tavanını aşıyor")
        return v

def validate_and_execute_tool(tool_name: str, raw_args: dict, schema: type[BaseModel]):
    try:
        validated = schema.model_validate(raw_args)
    except ValidationError as e:
        return {"status": "validation_error", "detail": e.errors()}
    return execute_tool(tool_name, validated)

Anti-pattern’ler

2.8 Human Escalation

Prensip: İnsan-döngüde (human-in-the-loop) otomasyonun bir başarısızlığı değildir — diğer herhangi bir fallback kadar bilinçli tasarlanması gereken tasarlanmış bir güvenilirlik katmanıdır.

Best practices

Design pattern: Tam context devri ile escalation

from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class EscalationTicket:
    session_id: str
    reason: str
    confidence: float | None
    conversation: list[dict]
    agent_plan: list[str]
    tool_results: dict
    created_at: datetime = field(default_factory=datetime.utcnow)
    priority: str = "normal"

class EscalationQueue:
    def __init__(self, backend):
        self.backend = backend

    def escalate(self, checkpoint: AgentCheckpoint, reason: str, confidence: float | None = None):
        priority = "high" if reason in {"guardrail_violation", "financial_high_risk"} else "normal"
        ticket = EscalationTicket(
            session_id=checkpoint.session_id,
            reason=reason,
            confidence=confidence,
            conversation=checkpoint.tool_results.get("conversation", []),
            agent_plan=checkpoint.plan,
            tool_results=checkpoint.tool_results,
            priority=priority,
        )
        self.backend.enqueue(ticket)
        return ticket

def should_escalate(confidence: float, threshold: float = 0.6) -> bool:
    return confidence < threshold

Anti-pattern’ler

3. Evaluation

Prensip: “Çalışıyor” bir metrik değil. Bir agent sistemi, ship edilmeden önce, ilk olaydan sonra değil, ölçülebilir, tekrarlanabilir, versiyonlanmış bir evaluation harness’ine ihtiyaç duyar.

3.1 Ne ölçülmeli

                     Agent
                       │
               ┌───────┴───────┐
               ▼               ▼
            Başarı           Başarısızlık
               │                │
        ┌──────┼──────┐    ┌────┴────┐
        ▼      ▼      ▼    ▼         ▼
      kalite  latency maliyet  tür   sıklık

3.2 Best practices

Design pattern: Eval harness iskeleti

from dataclasses import dataclass
from typing import Callable, Any

@dataclass
class EvalCase:
    id: str
    input: Any
    expected: Any
    tags: list[str]

@dataclass
class EvalResult:
    case_id: str
    passed: bool
    score: float
    latency_ms: float
    cost_tokens: int
    failure_mode: str | None = None

class EvalHarness:
    def __init__(self, cases: list[EvalCase], scorer: Callable[[Any, Any], float],
                 pass_threshold: float = 0.8):
        self.cases = cases
        self.scorer = scorer
        self.pass_threshold = pass_threshold

    def run(self, agent_fn: Callable[[Any], Any]) -> list[EvalResult]:
        results = []
        for case in self.cases:
            start = time.monotonic()
            try:
                output = agent_fn(case.input)
                score = self.scorer(output, case.expected)
                results.append(EvalResult(
                    case_id=case.id,
                    passed=score >= self.pass_threshold,
                    score=score,
                    latency_ms=(time.monotonic() - start) * 1000,
                    cost_tokens=getattr(output, "token_usage", 0),
                ))
            except Exception as e:
                results.append(EvalResult(
                    case_id=case.id, passed=False, score=0.0,
                    latency_ms=(time.monotonic() - start) * 1000,
                    cost_tokens=0, failure_mode=type(e).__name__,
                ))
        return results

    def summary(self, results: list[EvalResult]) -> dict:
        n = len(results)
        return {
            "pass_rate": sum(r.passed for r in results) / n,
            "avg_score": sum(r.score for r in results) / n,
            "p90_latency_ms": sorted(r.latency_ms for r in results)[int(n * 0.9)],
            "total_cost_tokens": sum(r.cost_tokens for r in results),
            "failure_modes": {r.failure_mode for r in results if r.failure_mode},
        }

Anti-pattern’ler

4. Observability

Prensip: Herhangi bir tek istek için sonradan tam olarak ne olduğunu — her prompt’u, her tool çağrısını, her model yanıtını, her retry’ı — yeniden inşa edemiyorsan, bir production agent sistemine değil, bir UI’lı kara kutuya sahipsin.

4.1 Trace hiyerarşisi

request
  ↓
graph            (bu istek için genel agent workflow'u / DAG)
  ↓
node             (graph içindeki bir adım — plan, retrieve, generate, act)
  ↓
agent            (multi-agent ise belirli bir alt-agent veya rol)
  ↓
tool             (bireysel tool çağrısı)
  ↓
LLM              (bireysel model API çağrısı — prompt, yanıt, token, latency)

4.2 Best practices

Design pattern: Hiyerarşik tracing

from contextvars import ContextVar
from dataclasses import dataclass, field
from datetime import datetime
import uuid

current_trace: ContextVar[str] = ContextVar("current_trace", default="")

@dataclass
class Span:
    span_id: str
    parent_id: str | None
    trace_id: str
    layer: str  # request | graph | node | agent | tool | llm
    name: str
    start: datetime = field(default_factory=datetime.utcnow)
    end: datetime | None = None
    metadata: dict = field(default_factory=dict)

class Tracer:
    """Minimal hiyerarşik tracer — production'da backend'i
    OpenTelemetry/Honeycomb/Datadog ile değiştir."""

    def __init__(self, sink: Callable[[Span], None]):
        self.sink = sink
        self._stack: list[Span] = []

    def start_span(self, layer: str, name: str, **metadata) -> Span:
        trace_id = current_trace.get() or str(uuid.uuid4())
        current_trace.set(trace_id)
        parent = self._stack[-1].span_id if self._stack else None
        span = Span(str(uuid.uuid4()), parent, trace_id, layer, name, metadata=metadata)
        self._stack.append(span)
        return span

    def end_span(self, span: Span, **result_metadata):
        span.end = datetime.utcnow()
        span.metadata.update(result_metadata)
        self._stack.pop()
        self.sink(span)  # logging/tracing backend'e gönder

# Kullanım
tracer = Tracer(sink=lambda s: logger.info("span", **s.__dict__))

def call_llm_traced(tracer, prompt, model):
    span = tracer.start_span("llm", model, prompt_preview=prompt[:200])
    response = client.messages.create(model=model, messages=[{"role": "user", "content": prompt}], max_tokens=1024)
    tracer.end_span(span,
        input_tokens=response.usage.input_tokens,
        output_tokens=response.usage.output_tokens,
        cached_tokens=getattr(response.usage, "cache_read_input_tokens", 0),
    )
    return response

Anti-pattern’ler

5. Hepsini Bir Araya Getirmek

Production seviyesinde bir agent isteği, prensipte, yukarıdaki her katmandan geçmelidir:

Kullanıcı isteği
   │
   ▼
[Guardrail: input kontrolü] ──başarısız──► reddet / sanitize et
   │ geçti
   ▼
[Context Assembler: bütçelenmiş, sıralanmış retrieval]
   │
   ▼
[Model Router: görev karmaşıklığına göre katman seç]
   │
   ▼
[Deadline-sınırlı agent döngüsü]
   │
   ├─► [Tool çağrısı] ──idempotency key──► [Validation] ──► çalıştır ──► [Trace: tool span]
   │        │
   │        └─başarısız──► [Retry politikası] ──tükendi──► [Fallback zinciri]
   │
   ▼
[Structured output üretimi] ──validasyon başarısız──► [Self-repair döngüsü] ──tükendi──► [Escalation]
   │ geçti
   ▼
[Guardrail: output kontrolü] ──başarısız──► [Escalation]
   │ geçti
   ▼
[Checkpoint durumu kalıcı hale getirildi] ──► yanıt kullanıcıya stream edildi
   │
   ▼
[Tam trace gönderildi: request → graph → node → agent → tool → LLM]
   │
   ▼
[Sonuç kaydedildi] ──► [Eval dataset / regresyon suite güncellendi]

Bu dokümanın tek cümlelik özeti: bir agent, bir kere doğru bir cevap ürettiğinde “bitmiş” değildir — güvenilir, ucuz, gözlemlenebilir ve kurtarılabilir şekilde doğru bir cevabı, yaptığı her dış çağrının er ya da geç başarısız olacağı varsayımı altında ürettiğinde bitmiştir.