AI Agents · 课时

使用 LangSmith 和 Langfuse 分析追踪记录

阅读追踪记录:识别缓慢的工具、错误的决策和错误模式

第 1 / 4 课13 个步骤

使用 LangSmith 和 Langfuse 分析追踪记录 是 CoddyKit 上的免费 AI Agents 课时。 这是第 1 节课,共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 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 用户界面中筛选和搜索。对于跟踪不同的代理版本、用户 ID 或实验标签很有用。

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 框架或自定义代码。使用 Langfuse SDK 手动创建跟踪记录和 span。

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

在 Langfuse 中跟踪 LLM 调用

为每次 LLM 调用创建一个 generation span。这样可以捕获所使用的模型、提示词、补全内容和令牌计数——这些是成本分析最重要的数据。

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 适用于任何框架并让您拥有更多控制权。两者都会记录每个代理步骤的输入、输出、令牌使用量、延迟和错误。使用筛选、评分和警报来维护生产环境中的代理质量。

免费开始

用 AI 导师学习 AI Agents — 免费

在浏览器中编写并运行真实代码,获得全天候 AI 导师的即时帮助,并在网页或应用中继续学习。

课程
60
课程
239

常见问题解答

「使用 LangSmith 和 Langfuse 分析追踪记录」课时是免费的吗?

是的 — 「使用 LangSmith 和 Langfuse 分析追踪记录」的完整文本可在网页上免费阅读。要进行交互式练习(内置代码编辑器和全天候 AI 导师)并解锁 AI Agents 课程的其余内容,请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「使用 LangSmith 和 Langfuse 分析追踪记录」这节课中我会学到什么?

阅读追踪记录:识别缓慢的工具、错误的决策和错误模式 你通过在浏览器中直接运行的动手代码来练习 AI Agents,全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗?

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者,你可以从这里开始或从头开始,按照自己的节奏学习。 这是第 1 节课,共 4 节。

「使用 LangSmith 和 Langfuse 分析追踪记录」课时需要多长时间?

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动,所以你能稳步进步,并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗?

能。每节 AI Agents 课都包含内置代码编辑器,你可以在浏览器中直接编写并运行真实代码,并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. 使用 LangSmith 和 Langfuse 分析追踪记录
  2. 逐步分析令牌消耗与成本
  3. 识别缓慢且昂贵的步骤
  4. 分析智能体故障的根本原因
← 返回 AI Agents