使用 LangSmith 和 Langfuse 分析追踪记录
阅读追踪记录:识别缓慢的工具、错误的决策和错误模式
使用 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 反馈 — 无需本地设置。
此课程中的所有课时
- 使用 LangSmith 和 Langfuse 分析追踪记录
- 逐步分析令牌消耗与成本
- 识别缓慢且昂贵的步骤
- 分析智能体故障的根本原因