0Pricing
AI Agents · Урок

Анализ трассировок с LangSmith и Langfuse

Чтение трассировок: выявление медленных инструментов, неверных решений и закономерностей ошибок.

«Анализ трассировок с LangSmith и Langfuse» — бесплатный урок AI Agents на CoddyKit. Это урок 1 из 4. Ты можешь прочитать весь урок бесплатно ниже — а потом практиковать его прямо в браузере с встроенным редактором кода и ИИ-репетитором 24/7. Это часть пути обучения AI Agents, и твой прогресс синхронизируется между веб-версией и приложением CoddyKit. Курс AI Agents содержит 4 уроков всего.

Зачем трассировать работу агента

Агенты выполняют несколько вызовов LLM и инструментов за один запуск. Без трассировки отладка превращается в угадывание. Трассировка записывает каждый шаг: входные данные, выходные данные, использование токенов, задержку и ошибки — предоставляя полную картину каждого запуска.

Настройка LangSmith

LangSmith — платформа Anthropic для трассировки LangChain. Включите её, задав две переменные окружения. Каждый вызов LangChain автоматически трассируется и становится доступен в интерфейсе 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

Добавление метаданных запуска

Добавляйте метки и метаданные к трассировкам, чтобы фильтровать и искать их в интерфейсе LangSmith. Это полезно для отслеживания разных версий агента, идентификаторов пользователей или меток экспериментов.

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

На панели 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 для пользовательской трассировки

Langfuse — альтернатива LangSmith с открытым исходным кодом. Она работает с любым фреймворком для LLM или пользовательским кодом. Используйте SDK 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

Создавайте интервал generation для каждого вызова LLM. Он фиксирует использованную модель, запрос, ответ и количество токенов — наиболее важные данные для анализа затрат.

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)

Фильтрация запусков по ошибкам и задержке

Используйте клиент LangSmith, чтобы программно находить проблемные запуски. Фильтруйте их по статусу ошибки, порогу задержки или определённым меткам, чтобы сосредоточить усилия на отладке.

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)')

Сравнение запусков

LangSmith позволяет сравнивать два запуска в своём интерфейсе и видеть, что изменилось. Программно вы можете сравнивать выходные данные запусков, использование токенов и задержку, чтобы выявлять регрессии после изменений модели или запроса.

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")')

Структурированный контекст трассировки

Добавляйте к трассировкам содержательный контекст: ID сеанса, ID пользователя, версию агента и флаги возможностей. Благодаря этому легко разделять трассировки на группы и сравнивать производительность при разных конфигурациях.

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)

Настройка оповещений

Следите за состоянием агента, настроив оповещения в LangSmith или Langfuse. Настройте оповещение, если частота ошибок превышает порог, задержка P99 резко возрастает или определённый шаг постоянно завершается с ошибкой.

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')

Проверка знаний: трассировка

Проверьте своё понимание трассировки агентов с помощью LangSmith и Langfuse.

Обзор трассировки

LangSmith и Langfuse дополняют друг друга: LangSmith тесно интегрируется с LangChain и требует минимальной настройки, а Langfuse работает с любым фреймворком и предоставляет больше контроля. Оба инструмента записывают входные данные, выходные данные, использование токенов, задержку и ошибки для каждого шага агента. Используйте фильтрацию, оценки и оповещения, чтобы поддерживать качество агента в рабочей среде.

Часто задаваемые вопросы

Урок «Анализ трассировок с LangSmith и Langfuse» бесплатный?

Да — полный текст урока «Анализ трассировок с LangSmith и Langfuse» бесплатно доступен здесь в веб-версии. Чтобы практиковать его интерактивно (встроенный редактор кода и ИИ-репетитор 24/7) и разблокировать остальной курс AI Agents, подпишись на CoddyKit PRO. Курс AI Agents содержит 4 уроков всего.

Чему я научусь в уроке «Анализ трассировок с LangSmith и Langfuse»?

Чтение трассировок: выявление медленных инструментов, неверных решений и закономерностей ошибок. Ты практикуешь AI Agents с помощью реального кода, который запускаешь прямо в браузере, и ИИ-репетитор 24/7 отвечает на твои вопросы во время урока.

Нужен ли мне опыт, чтобы начать AI Agents?

Предыдущий опыт не требуется. AI Agents на CoddyKit структурирован для всех уровней — от новичков до продвинутых, поэтому ты можешь начать отсюда или с самого начала и учиться в своем темпе. Это урок 1 из 4.

Сколько времени занимает урок «Анализ трассировок с LangSmith и Langfuse»?

Большинство уроков CoddyKit занимают около 5–10 минут. Каждый из них компактный и интерактивный, поэтому ты постоянно делаешь прогресс и продолжаешь с того же места в веб-версии и приложении.

Можно ли писать и запускать код в этом уроке AI Agents?

Да. Каждый урок AI Agents включает встроенный редактор кода, поэтому ты пишешь и запускаешь реальный код прямо в браузере и получаешь моментальную обратную связь от AI — локальная установка не требуется.

Все уроки этого курса

  1. Анализ трассировок с LangSmith и Langfuse
  2. Профилирование токенов и затрат по шагам
  3. Выявление медленных и затратных шагов
  4. Анализ первопричин сбоев агента
← Назад к AI Agents