LangSmith SDK 可观测性

import os

# 必需：LangSmith API密钥
os.environ["LANGSMITH_API_KEY"] = "lsv2_pt_..."

# 必需：项目名称（用于组织追踪数据）
os.environ["LANGSMITH_PROJECT"] = "my-llm-project"

# 可选：启用/禁用追踪
os.environ["LANGCHAIN_TRACING_V2"] = "true"  # 启用追踪

# 可选：LangSmith端点（自托管时修改）
os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com"

import langsmith
print(langsmith.__version__)

# 验证API连接
from langsmith import Client
client = Client()
print(client.list_projects())

import os

# === 核心配置 ===
os.environ["LANGSMITH_API_KEY"] = "lsv2_pt_..."          # API密钥
os.environ["LANGSMITH_PROJECT"] = "my-project"            # 项目名称

# === 追踪配置 ===
os.environ["LANGCHAIN_TRACING_V2"] = "true"               # 启用v2追踪
os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com"  # API端点

# === 高级配置 ===
os.environ["LANGSMITH_HIDE_INPUTS"] = "false"             # 是否隐藏输入（隐私保护）
os.environ["LANGSMITH_HIDE_OUTPUTS"] = "false"            # 是否隐藏输出
os.environ["LANGSMITH_HIDE_SECRETS"] = "true"             # 自动隐藏敏感信息
os.environ["LANGSMITH_SAMPLE_RATE"] = "1.0"               # 采样率（0.0-1.0）
os.environ["LANGSMITH_TRACING"] = "true"                  # 启用追踪的另一种方式

from langsmith import Client

client = Client(
    api_url="https://api.smith.langchain.com",
    api_key="lsv2_pt_...",
    timeout=30
)

from langsmith import traceable

@traceable(
    name="my_function",           # 可选：追踪中显示的名称，默认为函数名
    run_type="chain",             # 可选：运行类型，如"chain"、"llm"、"retriever"、"tool"
    metadata={"version": "1.0"},  # 可选：元数据字典
    tags=["production", "v2"],    # 可选：标签列表
    reduce_fn=None,               # 可选：用于聚合子追踪结果的函数
)
def my_function(input_text: str) -> str:
    # 函数逻辑
    return f"处理结果: {input_text}"

# 调用时自动记录追踪
result = my_function("你好")

from langsmith import traceable
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4")

@traceable(name="retriever", run_type="retriever")
def retrieve_documents(query: str) -> list:
    # 模拟文档检索
    return [f"文档1: {query}的相关信息", f"文档2: 更多关于{query}的内容"]

@traceable(name="generator", run_type="chain")
def generate_answer(query: str, documents: list) -> str:
    context = "\n".join(documents)
    response = llm.invoke(f"基于以下上下文回答问题：\n{context}\n\n问题：{query}")
    return response.content

@traceable(name="rag_pipeline", run_type="chain")
def rag_pipeline(query: str) -> str:
    # 嵌套追踪：rag_pipeline → retriever → generator
    docs = retrieve_documents(query)
    answer = generate_answer(query, docs)
    return answer

# 执行后，LangSmith中会看到完整的追踪树
result = rag_pipeline("什么是量子计算？")

from langsmith import traceable
from langsmith.run_helpers import get_current_run_tree

@traceable(name="manual_trace")
def my_function(input_text: str) -> str:
    # 获取当前追踪上下文
    run = get_current_run_tree()

    # 添加元数据
    run.metadata["custom_field"] = "custom_value"

    # 手动添加子追踪
    with run.new_child(
        name="sub_step",
        run_type="tool",
        inputs={"param": "value"}
    ) as child_run:
        result = do_something()
        child_run.outputs = {"result": result}

    return result

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 只要配置了环境变量，以下链的所有步骤会自动追踪
llm = ChatOpenAI(model="gpt-4")

prompt = ChatPromptTemplate.from_template("用中文解释：{topic}")
chain = prompt | llm | StrOutputParser()

# 调用chain时，LangSmith自动记录：
# - prompt的输入和格式化后的提示词
# - LLM的调用参数、输入、输出、token消耗
# - 解析器的输入和输出
result = chain.invoke({"topic": "量子计算"})

from langsmith.evaluation import evaluate

results = evaluate(
    target=target_function,           # 必需：被评估的目标函数
    data=dataset_name,                # 必需：数据集名称或数据集对象
    evaluators=[evaluator1, evaluator2], # 必需：评估器列表
    experiment_prefix="my-exp",       # 可选：实验名称前缀
    description="评估RAG系统质量",     # 可选：实验描述
    max_concurrency=5,                # 可选：最大并发数
    metadata={"version": "2.0"},      # 可选：元数据
)

from langsmith.evaluation import Evaluator, EvaluationResult

class RelevanceEvaluator(Evaluator):
    """自定义相关性评估器"""

    def __init__(self, llm=None):
        self.llm = llm or ChatOpenAI(model="gpt-4", temperature=0)

    def evaluate(self, *, run, example) -> EvaluationResult:
        # run: 目标函数的运行结果
        # example: 数据集中的样本（包含input和reference_output）
        input_question = example.inputs["question"]
        actual_output = run.outputs["output"]
        reference = example.outputs.get("answer", "")

        # 使用LLM评估相关性
        eval_prompt = f"""评估以下回答与问题的相关性。

问题：{input_question}
回答：{actual_output}
参考答案：{reference}

请给出0到1之间的相关性分数。"""

        response = self.llm.invoke(eval_prompt)
        score = self._parse_score(response.content)

        return EvaluationResult(
            score=score,                    # 0.0-1.0之间的分数
            comment=f"相关性评估: {response.content}",  # 评估评论
            key="relevance"                 # 指标键名
        )

# 使用自定义评估器
results = evaluate(
    target=rag_pipeline,
    data="my-qa-dataset",
    evaluators=[RelevanceEvaluator()]
)

from langsmith.evaluation import evaluate

def relevance_evaluator(run, example) -> dict:
    """简化的函数式评估器"""
    question = example.inputs["question"]
    answer = run.outputs["output"]

    # 简单的关键词匹配评估
    keywords = example.outputs.get("keywords", [])
    covered = sum(1 for kw in keywords if kw in answer)
    score = covered / len(keywords) if keywords else 0

    return {"score": score, "comment": f"覆盖了{covered}/{len(keywords)}个关键词"}

results = evaluate(
    target=rag_pipeline,
    data="my-qa-dataset",
    evaluators=[relevance_evaluator]
)

from langsmith.evaluation import evaluate
from langchain_openai import ChatOpenAI

# LangSmith提供了预构建的LLM评估器
from langsmith.evaluation.evaluators import (
    CriteriaEvaluator,
    LabeledCriteriaEvaluator,
    EmbeddingDistance,
    StringDistance
)

# 基于准则的评估（无需参考答案）
helpfulness_evaluator = CriteriaEvaluator(
    criteria="helpfulness",    # 内置准则：helpfulness, conciseness, correctness等
    llm=ChatOpenAI(model="gpt-4", temperature=0)
)

# 带标签的准则评估（需要参考答案）
correctness_evaluator = LabeledCriteriaEvaluator(
    criteria="correctness",
    llm=ChatOpenAI(model="gpt-4", temperature=0)
)

# 嵌入距离评估
embedding_evaluator = EmbeddingDistance(
    embeddings=OpenAIEmbeddings()
)

# 字符串距离评估
string_evaluator = StringDistance(
    distance="edit_distance"  # 编辑距离
)

results = evaluate(
    target=my_llm_function,
    data="test-dataset",
    evaluators=[
        helpfulness_evaluator,
        correctness_evaluator,
        embedding_evaluator
    ]
)

from langsmith import Client

client = Client()

# === 创建数据集 ===

# 方式1：从字典列表创建
dataset = client.create_dataset(
    dataset_name="qa-test-dataset",
    description="问答系统测试数据集"
)

# 添加样本
client.create_examples(
    dataset_id=dataset.id,
    inputs=[
        {"question": "什么是机器学习？"},
        {"question": "深度学习和机器学习的区别？"},
    ],
    outputs=[
        {"answer": "机器学习是让计算机从数据中自动学习的技术。", "keywords": ["机器学习", "数据", "自动学习"]},
        {"answer": "深度学习是机器学习的子集，使用多层神经网络。", "keywords": ["深度学习", "机器学习", "神经网络"]},
    ]
)

# 方式2：从CSV上传
client.upload_csv(
    dataset_name="csv-dataset",
    csv_file="test_data.csv",
    input_keys=["question"],         # 输入列
    output_keys=["answer", "keywords"] # 输出列
)

# === 查询数据集 ===

# 列出所有数据集
datasets = list(client.list_datasets())
for ds in datasets:
    print(f"数据集: {ds.name}, 样本数: {client.list_examples(dataset_id=ds.id).total_count}")

# 获取特定数据集
dataset = client.read_dataset(dataset_name="qa-test-dataset")

# 获取数据集中的样本
examples = list(client.list_examples(dataset_id=dataset.id))
for ex in examples:
    print(f"输入: {ex.inputs}, 输出: {ex.outputs}")

# === 更新和删除 ===

# 更新数据集
client.update_dataset(
    dataset_id=dataset.id,
    description="更新后的描述"
)

# 删除数据集
client.delete_dataset(dataset_name="old-dataset")

from langsmith import Client

client = Client(
    api_url="https://api.smith.langchain.com",  # API端点
    api_key="lsv2_pt_...",                       # API密钥（也可通过环境变量设置）
    timeout=30,                                  # 请求超时时间（秒）
    retry_max=2                                  # 最大重试次数
)

# === 追踪相关 ===

# 读取追踪数据
runs = list(client.list_runs(
    project_name="my-project",
    run_type="chain",
    error=True  # 仅查看出错的运行
))

# 获取特定运行的详情
run = client.read_run(run_id="abc-123")
print(f"运行名称: {run.name}")
print(f"输入: {run.inputs}")
print(f"输出: {run.outputs}")
print(f"耗时: {run.total_time}")
print(f"Token消耗: {run.total_tokens}")

# === 项目管理 ===

# 列出项目
projects = list(client.list_projects())

# 创建项目
project = client.create_project(
    project_name="new-project",
    description="新项目"
)

# === 评估结果 ===

# 读取评估结果
eval_results = list(client.list_evaluations(
    dataset_name="qa-test-dataset"
))

import os
from langsmith import traceable, Client
from langsmith.evaluation import evaluate
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

# 配置LangSmith
os.environ["LANGSMITH_API_KEY"] = "lsv2_pt_..."
os.environ["LANGSMITH_PROJECT"] = "rag-evaluation"

# === 构建带追踪的RAG管道 ===

llm = ChatOpenAI(model="gpt-4", temperature=0)
embeddings = OpenAIEmbeddings()

# 假设已有向量数据库
vectorstore = Chroma.from_texts(
    ["机器学习是AI的子领域", "深度学习使用神经网络"],
    embeddings
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

@traceable(name="retrieve", run_type="retriever")
def retrieve(query: str) -> list:
    docs = retriever.invoke(query)
    return [doc.page_content for doc in docs]

@traceable(name="generate", run_type="chain")
def generate(query: str, contexts: list) -> str:
    context_text = "\n".join(contexts)
    prompt = f"基于以下上下文回答问题：\n{context_text}\n\n问题：{query}"
    response = llm.invoke(prompt)
    return response.content

@traceable(name="rag_pipeline", run_type="chain")
def rag_pipeline(question: str) -> dict:
    contexts = retrieve(question)
    answer = generate(question, contexts)
    return {"answer": answer, "contexts": contexts}

# === 创建评估数据集 ===

client = Client()
dataset = client.create_dataset(
    dataset_name="rag-eval-dataset",
    description="RAG系统评估数据集"
)

client.create_examples(
    dataset_id=dataset.id,
    inputs=[
        {"question": "什么是机器学习？"},
        {"question": "深度学习和机器学习有什么区别？"},
    ],
    outputs=[
        {"answer": "机器学习是让计算机从数据中自动学习的AI技术。"},
        {"answer": "深度学习是机器学习的子集，核心是多层神经网络。"},
    ]
)

# === 定义评估器 ===

def answer_quality_evaluator(run, example) -> dict:
    """评估答案质量"""
    actual = run.outputs["answer"]
    expected = example.outputs["answer"]

    eval_llm = ChatOpenAI(model="gpt-4", temperature=0)
    prompt = f"""评估以下答案的质量。

问题：{example.inputs["question"]}
实际答案：{actual}
参考答案：{expected}

请从准确性和完整性两个维度评估，给出0-1之间的分数。"""

    response = eval_llm.invoke(prompt)
    score = 0.8  # 实际中应解析response
    return {"score": score}

def context_relevance_evaluator(run, example) -> dict:
    """评估检索上下文的相关性"""
    contexts = run.outputs.get("contexts", [])
    question = example.inputs["question"]

    if not contexts:
        return {"score": 0.0, "comment": "无检索上下文"}

    eval_llm = ChatOpenAI(model="gpt-4", temperature=0)
    prompt = f"""评估检索到的上下文与问题的相关性。

问题：{question}
检索上下文：{contexts}

给出0-1之间的相关性分数。"""

    response = eval_llm.invoke(prompt)
    score = 0.7  # 实际中应解析response
    return {"score": score}

# === 执行评估 ===

def target_function(inputs: dict) -> dict:
    return rag_pipeline(inputs["question"])

results = evaluate(
    target=target_function,
    data="rag-eval-dataset",
    evaluators=[answer_quality_evaluator, context_relevance_evaluator],
    experiment_prefix="rag-v1"
)

from langsmith import traceable
from langchain_openai import ChatOpenAI
from langchain.agents import create_openai_functions_agent, AgentExecutor
from langchain.tools import tool

# 定义带追踪的工具
@tool
@traceable(name="search_tool", run_type="tool")
def search_tool(query: str) -> str:
    """搜索相关信息"""
    return f"搜索结果：关于{query}的信息..."

@tool
@traceable(name="calculator", run_type="tool")
def calculator(expression: str) -> str:
    """计算数学表达式"""
    try:
        return str(eval(expression))
    except Exception as e:
        return f"计算错误：{e}"

# 创建Agent
# Agent的每一步思考、工具调用都会被自动追踪
llm = ChatOpenAI(model="gpt-4", temperature=0)
tools = [search_tool, calculator]
# agent = create_openai_functions_agent(llm, tools)
# agent_executor = AgentExecutor(agent=agent, tools=tools)

# 调用Agent时，LangSmith会记录：
# - Agent的每一步推理
# - 工具调用的输入输出
# - 最终答案
# result = agent_executor.invoke({"input": "计算2的10次方加上3的5次方"})

from langsmith import traceable
from langsmith.evaluation import evaluate
from langchain_openai import ChatOpenAI

# 定义不同版本的提示词
prompt_v1 = """请回答以下问题：{question}"""

prompt_v2 = """你是一个专业的AI助手。请基于你的知识，详细、准确地回答以下问题：
{question}

要求：
1. 答案要准确无误
2. 尽量提供具体的细节和例子
3. 如果不确定，请明确说明"""

@traceable(name="llm_v1")
def answer_v1(inputs: dict) -> dict:
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    response = llm.invoke(prompt_v1.format(question=inputs["question"]))
    return {"answer": response.content}

@traceable(name="llm_v2")
def answer_v2(inputs: dict) -> dict:
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    response = llm.invoke(prompt_v2.format(question=inputs["question"]))
    return {"answer": response.content}

# 对比评估
def quality_evaluator(run, example) -> dict:
    answer = run.outputs["answer"]
    reference = example.outputs.get("answer", "")
    # 评估逻辑...
    return {"score": 0.8}

results_v1 = evaluate(
    target=answer_v1,
    data="qa-dataset",
    evaluators=[quality_evaluator],
    experiment_prefix="prompt-v1"
)

results_v2 = evaluate(
    target=answer_v2,
    data="qa-dataset",
    evaluators=[quality_evaluator],
    experiment_prefix="prompt-v2"
)

from langsmith import traceable, Client
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4")

@traceable(
    name="production_chat",
    run_type="chain",
    metadata={"environment": "production", "version": "2.1.0"},
    tags=["production", "chat-api"]
)
def chat(user_input: str, user_id: str) -> str:
    response = llm.invoke(user_input)
    return response.content

# 在生产API中使用
def handle_request(user_input: str, user_id: str):
    result = chat(user_input, user_id)
    return result

# 事后分析追踪数据
client = Client()

# 查看最近的错误
error_runs = list(client.list_runs(
    project_name="production-project",
    error=True,
    limit=10
))

for run in error_runs:
    print(f"运行: {run.name}, 错误: {run.error}, 时间: {run.start_time}")

# 分析延迟
recent_runs = list(client.list_runs(
    project_name="production-project",
    limit=100
))

latencies = [run.total_time for run in recent_runs if run.total_time]
avg_latency = sum(latencies) / len(latencies) if latencies else 0
print(f"平均延迟: {avg_latency:.2f}秒")

# 分析token消耗
total_tokens = sum(run.total_tokens or 0 for run in recent_runs)
print(f"总token消耗: {total_tokens}")

# 简化的实现原理
import functools
from contextvars import ContextVar

current_run: ContextVar = ContextVar("current_run", default=None)

def traceable(name=None, run_type="chain", **kwargs):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs_inner):
            parent_run = current_run.get()

            # 创建新的RunTree
            run = RunTree(
                name=name or func.__name__,
                run_type=run_type,
                inputs=extract_inputs(args, kwargs_inner),
                parent=parent_run
            )

            # 设置为当前追踪上下文
            token = current_run.set(run)

            try:
                # 执行原函数
                result = func(*args, **kwargs_inner)
                run.outputs = result
                run.end_time = datetime.now()
                return result
            except Exception as e:
                run.error = str(e)
                raise
            finally:
                # 恢复父追踪上下文
                current_run.reset(token)
                # 异步发送追踪数据
                send_run_async(run)

        return wrapper
    return decorator

# 大规模生产环境中，追踪可能影响性能
# 使用采样率控制追踪数据量
os.environ["LANGSMITH_SAMPLE_RATE"] = "0.1"  # 仅追踪10%的请求

# 对非关键路径禁用追踪
from langsmith import traceable

@traceable(name="non_critical_step", run_type="tool")
def non_critical_step():
    # 不影响核心业务逻辑的步骤
    pass

# 关闭追踪（调试完成后）
os.environ["LANGCHAIN_TRACING_V2"] = "false"

# 隐藏输入输出中的敏感信息
os.environ["LANGSMITH_HIDE_INPUTS"] = "true"
os.environ["LANGSMITH_HIDE_OUTPUTS"] = "true"

# 自动隐藏密钥等敏感信息
os.environ["LANGSMITH_HIDE_SECRETS"] = "true"

# 手动过滤敏感字段
@traceable(name="safe_pipeline")
def safe_pipeline(user_input: str) -> str:
    # 在追踪中只记录脱敏后的数据
    safe_input = user_input.replace(os.getenv("API_KEY", ""), "***")
    # ... 处理逻辑
    return result

# 按环境和功能组织项目
os.environ["LANGSMITH_PROJECT"] = "prod-customer-support"  # 生产环境
os.environ["LANGSMITH_PROJECT"] = "dev-experiment-v2"      # 开发实验

# 按版本追踪
@traceable(
    name="chat_api",
    metadata={"version": "2.1.0", "model": "gpt-4"},
    tags=["v2", "production"]
)
def chat_api(input_text):
    pass

# 检查配置
import os
print(os.environ.get("LANGCHAIN_TRACING_V2"))  # 应为"true"
print(os.environ.get("LANGSMITH_API_KEY"))     # 应非空
print(os.environ.get("LANGSMITH_PROJECT"))     # 应为有效项目名

# 确保LangChain版本兼容
# pip install --upgrade langchain langsmith

# 确保target函数的输入格式与数据集的inputs字段匹配
# 数据集inputs: {"question": "..."}
# target函数应接收: def target(inputs: dict) -> dict:
#     question = inputs["question"]  # 从inputs中提取question字段
#     ...
#     return {"answer": result}  # 返回字典

# 对内部实现细节不需要追踪的函数，不添加@traceable
# 仅对关键的入口函数和重要步骤添加追踪

@traceable(name="rag_pipeline")  # 追踪入口
def rag_pipeline(query):
    docs = retrieve_docs(query)      # 不追踪
    answer = generate_answer(query, docs)  # 不追踪
    return answer

# 而不是对每个内部函数都添加追踪