将追踪记录到特定项目

import os
import contextlib
from langsmith import Client, traceable, tracing_context

# 具有访问多个工作区权限的 API 密钥
api_key = os.getenv("LS_CROSS_WORKSPACE_KEY")

# 为不同工作区初始化客户端
workspace_a_client = Client(
    api_key=api_key,
    api_url="https://api.smith.langchain.com",
    workspace_id="<YOUR_WORKSPACE_A_ID>"  # 例如 "abc123..."
)

workspace_b_client = Client(
    api_key=api_key,
    api_url="https://api.smith.langchain.com",
    workspace_id="<YOUR_WORKSPACE_B_ID>"  # 例如 "def456..."
)

# 示例：基于客户 ID 进行路由
def get_workspace_client(customer_id: str):
    """根据客户路由到相应的工作区。"""
    if customer_id.startswith("premium_"):
        return workspace_a_client, "premium-customer-traces"
    else:
        return workspace_b_client, "standard-customer-traces"

@traceable
def process_request(data: dict, customer_id: str):
    """使用特定于工作区的追踪处理客户请求。"""
    # 您的业务逻辑在此处
    return {"status": "success", "data": data}

# 使用 tracing_context 路由到相应的工作区
def handle_customer_request(customer_id: str, request_data: dict):
    client, project_name = get_workspace_client(customer_id)

    # 此上下文内的所有内容都将被追踪到选定的工作区
    with tracing_context(enabled=True, client=client, project_name=project_name):
        result = process_request(request_data, customer_id)

    return result

# 示例用法
handle_customer_request("premium_user_123", {"query": "Hello"})
handle_customer_request("standard_user_456", {"query": "Hi"})

import os
import contextlib
from typing_extensions import TypedDict
from langgraph.graph import StateGraph
from langgraph.graph.state import RunnableConfig
from langsmith import Client, tracing_context

# 具有访问多个工作区权限的 API 密钥
api_key = os.getenv("LS_CROSS_WORKSPACE_KEY")

# 为不同工作区初始化客户端
workspace_a_client = Client(
    api_key=api_key,
    api_url="https://api.smith.langchain.com",
    workspace_id="<YOUR_WORKSPACE_A_ID>"
)

workspace_b_client = Client(
    api_key=api_key,
    api_url="https://api.smith.langchain.com",
    workspace_id="<YOUR_WORKSPACE_B_ID>"
)

# 定义用于工作区路由的配置模式
class Configuration(TypedDict):
    workspace_id: str

# 定义图状态
class State(TypedDict):
    response: str

def greeting(state: State, config: RunnableConfig) -> State:
    """生成特定于工作区的问候语。"""
    workspace_id = config.get("configurable", {}).get("workspace_id", "workspace_a")

    if workspace_id == "workspace_a":
        response = "Hello from Workspace A!"
    elif workspace_id == "workspace_b":
        response = "Hello from Workspace B!"
    else:
        response = "Hello from the default workspace!"

    return {"response": response}

# 构建基础图
base_graph = (
    StateGraph(state_schema=State, config_schema=Configuration)
    .add_node("greeting", greeting)
    .set_entry_point("greeting")
    .set_finish_point("greeting")
    .compile()
)

@contextlib.asynccontextmanager
async def graph(config):
    """根据配置将追踪动态路由到不同工作区。"""
    # 从配置中提取 workspace_id
    workspace_id = config.get("configurable", {}).get("workspace_id", "workspace_a")

    # 路由到相应的工作区
    if workspace_id == "workspace_a":
        client = workspace_a_client
        project_name = "production-traces"
    elif workspace_id == "workspace_b":
        client = workspace_b_client
        project_name = "development-traces"
    else:
        client = workspace_a_client
        project_name = "default-traces"

    # 为选定的工作区应用追踪上下文
    with tracing_context(enabled=True, client=client, project_name=project_name):
        yield base_graph

# 用法：使用不同的工作区配置进行调用
# await graph({"configurable": {"workspace_id": "workspace_a"}})
# await graph({"configurable": {"workspace_id": "workspace_b"}})

from langsmith import traceable, tracing_context
from langsmith.run_trees import WriteReplica, ApiKeyAuth

@traceable
def my_pipeline(query: str) -> str:
    # 您的应用程序逻辑在此处
    return f"Answer to: {query}"

replicas = [
    WriteReplica(
        api_url="https://api.smith.langchain.com",
        auth=ApiKeyAuth(api_key="ls__key_workspace_a"),
        project_name="project-prod",
    ),
    WriteReplica(
        api_url="https://api.smith.langchain.com",
        auth=ApiKeyAuth(api_key="ls__key_workspace_b"),
        project_name="project-staging",
        # 可选：覆盖复制运行上的字段
        updates={"metadata": {"environment": "staging"}},
    ),
]

with tracing_context(replicas=replicas):
    my_pipeline("What is LangSmith?")

from langsmith import traceable, tracing_context
from langsmith.run_trees import WriteReplica

@traceable
def my_pipeline(query: str) -> str:
    return f"Answer to: {query}"

with tracing_context(
    replicas=[
        WriteReplica(project_name="project-prod"),
        WriteReplica(project_name="project-staging", updates={"metadata": {"env": "staging"}}),
    ]
):
    my_pipeline("What is LangSmith?")