# llm_with_tracing.py
# 目的：具有完全可观察性的生产就绪 LLM 调用包装器。
# 在 Langfuse 中跟踪每个调用：输入、输出、令牌、成本、延迟。
#
# 先决条件：
# pip 安装 langfuse anthropic python-dotenv
#
＃ 设置：
# 1. 在 https://cloud.langfuse.com 创建一个免费帐户
# 2. 从“设置”>“API 密钥”获取密钥
# 3. 使用以下变量创建 .env 文件
#
＃ 跑步：
# python llm_with_tracing.py
进口 操作系统
进口 时间
从 多滕夫 进口 负载点环境
进口 人择的
从 朗福塞 进口 郎福斯
# 从.env文件加载环境变量
负载点环境（）
# .env 中所需的环境变量：
# LANGFUSE_PUBLIC_KEY=pk-lf-…
# LANGFUSE_SECRET_KEY=sk-lf-…
# LANGFUSE_HOST=https://cloud.langfuse.com （或您的自托管 URL）
# ANTHROPIC_API_KEY=sk-ant-…
# 初始化客户端
langfuse_client = 郎福斯（） # 从环境中自动读取密钥
人类客户端 = 人择的。人择（） # 从环境中读取 ANTHROPIC_API_KEY
# ── 配置 ──────────────────────────────────────────────────────────────
# 将您的提示存储在此处，而不是内嵌在 API 调用中。
# 这使得它可以独立版本化和测试。
系统提示符 = ””“你是一位乐于助人的客户支持助理。
清楚、简洁地回答问题。
如果你不知道什么，就直接说出来——不要猜测。”””
模型 = “克劳德十四行诗-4-20250514”
# Anthropic 截至 2026 年中期的定价（定价变化时更新）
# 用于计算每次调用的成本以进行成本跟踪
COST_PER_INPUT_TOKEN = 3.00 / 1_000_000 # 每百万输入代币 3.00 美元
COST_PER_OUTPUT_TOKEN = 15:00 / 1_000_000 # 每百万输出代币 15.00 美元
定义 call_llm_with_tracing（
用户消息: 斯特,
会话ID: 斯特 = “默认会话”,
用户身份: 斯特 = “匿名的”
） -> 斯特:
”””
拨打追踪的 LLM 电话。每次调用都会创建一个 Langfuse 跟踪：
– 完整的输入和输出
– 代币使用情况（输入、输出、总计）
– 以美元计算的成本
– 延迟（以毫秒为单位）
– 使用的模型和会话上下文
参数：
user_message ：来自用户的消息
session_id ：将相关呼叫分组到 Langfuse 中的一个对话中
user_id ：将呼叫与特定用户关联以进行分析
返回：
LLM 响应作为字符串
”””
# 为此用户交互创建顶级跟踪
# 跟踪作为一个工作单元出现在 Langfuse 仪表板中
痕迹 = langfuse_client。痕迹（
姓名=“客户支持电话”,
会话ID=会话ID,
用户身份=用户身份,
输入={“用户消息”: 用户消息, “系统提示”: 系统提示符}
）
# 在跟踪内创建一个生成跨度
# 这捕获特定于模型的详细信息：模型名称、令牌、成本
一代 = 痕迹。一代（
姓名=“克劳德完成”,
模型=模型,
输入={
“系统”: 系统提示符,
“消息”: [{“role”: “user”, “content”: user_message}]
}
）
开始时间 = 时间。时间（）
尝试:
# 进行API调用
回复 = 人类客户端。消息。创造（
模型=模型,
最大令牌数=1024,
系统=系统提示符,
消息=[{“role”: “user”, “content”: user_message}]
）
延迟时间 = 整数（（时间。时间（） – 开始时间） * 1000）
# 提取响应文本
响应文本 = 回复。内容[0]。文本
# 从响应中提取令牌使用情况
输入令牌 = 回复。用法。输入令牌
输出令牌 = 回复。用法。输出令牌
总令牌数 = 输入令牌 + 输出_代币
# 计算本次调用的成本
成本_美元 = （
输入令牌 * COST_PER_INPUT_TOKEN +
输出令牌 * 每输出成本_代币
）
# 用结果更新生成跨度
# 此数据填充 Langfuse 成本和代币仪表板
一代。结尾（
输出=响应文本,
用法={
“输入”: 输入令牌,
“输出”: 输出令牌,
“全部的”: 总令牌数,
“单元”: “代币”
},
元数据={
“延迟_毫秒”: 延迟时间,
“成本_美元”: 圆形的（成本_美元, 6）,
“模型”: 模型
}
）
# 使用最终输出更新跟踪
痕迹。更新（
输出={“回复”: 响应文本},
元数据={“总成本美元”: 圆形的（成本_美元, 6）}
）
# 将摘要打印到标准输出以供本地可见
打印（f“\n{‘─’ * 60}”）
打印（f“用户：{user_message}”）
打印（f“克劳德：{response_text}”）
打印（f“令牌：{input_tokens} 输入 / {output_tokens} 输出 / {total_tokens} 总计”）
打印（f“成本：${cost_usd:.6f}”）
打印（f“延迟：{latency_ms}ms”）
打印（f“跟踪：{langfuse_client.base_url}/trace/{trace.id}”）
打印（f“{‘─’ * 60}\n”）
返回 响应文本
除了 例外 作为 e:
# 在跟踪中记录错误，以便它显示在 Langfuse 中
一代。结尾（
输出=没有任何,
元数据={“错误”: 斯特（e）, “延迟_毫秒”: 整数（（时间。时间（） – 开始时间） * 1000）}
）
痕迹。更新（输出={“错误”: 斯特（e）}）
# 始终在引发之前刷新——确保发送错误跟踪
langfuse_client。冲水（）
增加
最后:
# Flush 将所有缓冲的事件发送到 Langfuse
# 在长时间运行的服务中，Langfuse 会自动刷新。
# 在脚本中，必须在进程退出之前手动刷新。
langfuse_client。冲水（）
# ── 运行演示 ────────────────────────────────────────────────────────
如果 __姓名__ == “__主要的__”:
# 模拟两轮客户支持对话
测试消息 = [
“What is your return policy for electronics?”,
“Can I return an item I bought 45 days ago?”
]
会议 = “演示会话-001”
为了 我, 信息 在 枚举（测试消息）:
打印（f“\n调用 {i + 1}/{len(test_messages)}”）
尝试:
call_llm_with_tracing（
用户消息=信息,
会话ID=会议,
用户身份=“测试用户42”
）
除了 例外 作为 e:
打印（f“调用 {i + 1} 时出错：{e}”）
