การวิเคราะห์ร่องรอยด้วย LangSmith และ Langfuse

เหตุใดจึงควรติดตามร่องรอยของเอเจนต์

การตั้งค่า LangSmith

import os
from dotenv import load_dotenv

load_dotenv()

# LangSmith tracing configuration
os.environ['LANGCHAIN_TRACING_V2'] = 'true'
os.environ['LANGCHAIN_API_KEY'] = os.environ.get('LANGSMITH_API_KEY', 'ls__...')
os.environ['LANGCHAIN_PROJECT'] = 'my-agent-project'

# Now any LangChain code is automatically traced
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(model='gpt-4o-mini', api_key=os.environ.get('OPENAI_API_KEY', 'sk-...'))

# This call is traced automatically
response = llm.invoke([HumanMessage(content='What is 2+2?')])
print(response.content)
# Check trace at: https://smith.langchain.com

การเพิ่มข้อมูลเมตาของการทำงาน

import os
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langsmith import traceable

os.environ['LANGCHAIN_TRACING_V2'] = 'true'
os.environ['LANGCHAIN_API_KEY'] = 'ls__your-key'
os.environ['LANGCHAIN_PROJECT'] = 'my-agent-project'

llm = ChatOpenAI(model='gpt-4o-mini', api_key='sk-...')

@traceable(name='my-agent-run', tags=['production', 'v2'], metadata={'user_id': '42'})
def run_agent(question: str) -> str:
    response = llm.invoke(
        [HumanMessage(content=question)],
        config={
            'run_name': f'agent-{question[:20]}',
            'tags': ['production'],
            'metadata': {'user_id': '42', 'version': 'v2.1'}
        }
    )
    return response.content

result = run_agent('Explain LangChain tracing')
print(result)

การดูร่องรอยในส่วนติดต่อผู้ใช้ของ LangSmith

# Programmatically query LangSmith for run data
from langsmith import Client

client = Client(api_key='ls__your-key')

# List recent runs for a project
runs = list(client.list_runs(
    project_name='my-agent-project',
    execution_order=1,      # Top-level runs only
    error=True,             # Only failed runs
    limit=10
))

for run in runs:
    print(f'Run: {run.name}')
    print(f'  Status: {run.status}')
    print(f'  Latency: {run.end_time - run.start_time if run.end_time else "running"}')
    print(f'  Error: {run.error}')
    print()

Langfuse สำหรับการติดตามร่องรอยแบบกำหนดเอง

from langfuse import Langfuse

lf = Langfuse(
    public_key='pk-lf-...',
    secret_key='sk-lf-...',
    host='https://cloud.langfuse.com'  # Or your self-hosted URL
)

# Create a trace
trace = lf.trace(
    name='email-agent-run',
    user_id='user-42',
    metadata={'environment': 'production'}
)

# Create a span for entity extraction
span = trace.span(
    name='entity-extraction',
    input={'text': 'Meeting with Alice from Google tomorrow'}
)

# Simulate work
extracted = ['Alice', 'Google']

# End the span with output
span.end(output={'entities': extracted})

print('Trace created in Langfuse')
print(f'View at: https://cloud.langfuse.com/trace/{trace.id}')

การติดตามการเรียกใช้ LLM ใน Langfuse

from langfuse import Langfuse
import openai

lf = Langfuse(public_key='pk-lf-...', secret_key='sk-lf-...')
client = openai.OpenAI(api_key='sk-...')

def traced_llm_call(trace, prompt: str, model: str = 'gpt-4o-mini') -> str:
    generation = trace.generation(
        name='llm-call',
        model=model,
        input=[{'role': 'user', 'content': prompt}]
    )
    
    response = client.chat.completions.create(
        model=model,
        messages=[{'role': 'user', 'content': prompt}]
    )
    content = response.choices[0].message.content
    
    generation.end(
        output=content,
        usage={
            'prompt_tokens': response.usage.prompt_tokens,
            'completion_tokens': response.usage.completion_tokens,
            'total_tokens': response.usage.total_tokens
        }
    )
    return content

trace = lf.trace(name='test-trace')
result = traced_llm_call(trace, 'What is the capital of France?')
print('Result:', result)

การกรองการทำงานตาม Error และเวลาแฝง

from langsmith import Client
from datetime import datetime, timedelta

client = Client(api_key='ls__your-key')

def find_slow_runs(project: str, latency_threshold_ms: int = 10000):
    runs = list(client.list_runs(
        project_name=project,
        execution_order=1,
        start_time=datetime.utcnow() - timedelta(hours=24)
    ))
    
    slow_runs = []
    for run in runs:
        if run.end_time and run.start_time:
            duration_ms = (run.end_time - run.start_time).total_seconds() * 1000
            if duration_ms > latency_threshold_ms:
                slow_runs.append({
                    'id': str(run.id),
                    'name': run.name,
                    'duration_ms': round(duration_ms),
                    'tags': run.tags
                })
    
    slow_runs.sort(key=lambda x: x['duration_ms'], reverse=True)
    return slow_runs

print('Find slow runs function defined')
print('Usage: find_slow_runs("my-agent-project", latency_threshold_ms=5000)')

การเปรียบเทียบการทำงาน

from langsmith import Client

client = Client(api_key='ls__your-key')

def compare_runs(run_id_1: str, run_id_2: str) -> dict:
    run1 = client.read_run(run_id_1)
    run2 = client.read_run(run_id_2)
    
    def get_tokens(run):
        if run.total_tokens:
            return run.total_tokens
        return 0
    
    def get_latency_ms(run):
        if run.end_time and run.start_time:
            return (run.end_time - run.start_time).total_seconds() * 1000
        return 0
    
    return {
        'run1': {'id': run_id_1, 'tokens': get_tokens(run1), 'latency_ms': get_latency_ms(run1), 'status': run1.status},
        'run2': {'id': run_id_2, 'tokens': get_tokens(run2), 'latency_ms': get_latency_ms(run2), 'status': run2.status},
        'token_delta': get_tokens(run2) - get_tokens(run1),
        'latency_delta_ms': get_latency_ms(run2) - get_latency_ms(run1)
    }

print('Run comparison function defined')

การเพิ่มคะแนนและข้อเสนอแนะ

from langsmith import Client

client = Client(api_key='ls__your-key')

def score_run(run_id: str, score: float, reasoning: str = ''):
    # score: 0.0 (bad) to 1.0 (perfect)
    client.create_feedback(
        run_id=run_id,
        key='quality',
        score=score,
        comment=reasoning
    )

def auto_evaluate_run(run_id: str, expected_output: str, actual_output: str) -> float:
    # Simple heuristic: check if key terms from expected output are present
    expected_terms = set(expected_output.lower().split())
    actual_terms = set(actual_output.lower().split())
    overlap = len(expected_terms & actual_terms) / max(len(expected_terms), 1)
    score = min(1.0, overlap * 1.5)  # Normalize
    score_run(run_id, score, f'Term overlap: {overlap:.2f}')
    return score

print('Scoring functions defined')
print('Example: score_run("run-id-abc", 0.85, "Good answer but missing one detail")')

บริบทของร่องรอยแบบมีโครงสร้าง

import os
from langsmith import traceable
from langchain_core.runnables import RunnableConfig

def build_trace_config(user_id: str, session_id: str, version: str) -> dict:
    return {
        'metadata': {
            'user_id': user_id,
            'session_id': session_id,
            'agent_version': version,
            'environment': os.environ.get('ENV', 'development')
        },
        'tags': [version, os.environ.get('ENV', 'development')],
        'run_name': f'agent-{user_id[:8]}'
    }

@traceable
def run_agent_with_context(question: str, user_id: str, session_id: str):
    config = build_trace_config(user_id, session_id, 'v2.3')
    # Pass config to any LangChain component
    # llm.invoke([HumanMessage(content=question)], config=config)
    print(f'Running agent for user {user_id}, session {session_id}')
    return 'Answer here'

result = run_agent_with_context('Question', 'user-001', 'sess-xyz')
print(result)

การตั้งค่าการแจ้งเตือน

from langsmith import Client
from datetime import datetime, timedelta

client = Client(api_key='ls__your-key')

def check_error_rate(project: str, window_minutes: int = 60, threshold: float = 0.05) -> dict:
    runs = list(client.list_runs(
        project_name=project,
        execution_order=1,
        start_time=datetime.utcnow() - timedelta(minutes=window_minutes)
    ))
    
    if not runs:
        return {'error_rate': 0.0, 'alert': False}
    
    error_count = sum(1 for r in runs if r.status == 'error')
    error_rate = error_count / len(runs)
    
    if error_rate > threshold:
        print(f'ALERT: Error rate {error_rate:.1%} exceeds threshold {threshold:.1%}')
        # Send to Slack/PagerDuty here
    
    return {
        'total_runs': len(runs),
        'error_count': error_count,
        'error_rate': round(error_rate, 4),
        'alert': error_rate > threshold
    }

print('Error rate monitor defined')

แบบทดสอบความรู้: การติดตามร่องรอย

สรุปการติดตามร่องรอย