Log traces to a specific project

import openai
from langsmith import traceable
from langsmith.run_trees import RunTree

client = openai.Client()
messages = [
  {"role": "system", "content": "You are a helpful assistant."},
  {"role": "user", "content": "Hello!"}
]

# Use the @traceable decorator with the 'project_name' parameter to log traces to LangSmith
# Ensure that the LANGSMITH_TRACING environment variables is set for @traceable to work
@traceable(
  run_type="llm",
  name="OpenAI Call Decorator",
  project_name="My Project"
)
def call_openai(
  messages: list[dict], model: str = "gpt-4o-mini"
) -> str:
  return client.chat.completions.create(
      model=model,
      messages=messages,
  ).choices[0].message.content

# Call the decorated function
call_openai(messages)

# You can also specify the Project via the project_name parameter
# This will override the project_name specified in the @traceable decorator
call_openai(
  messages,
  langsmith_extra={"project_name": "My Overridden Project"},
)

# The wrapped OpenAI client accepts all the same langsmith_extra parameters
# as @traceable decorated functions, and logs traces to LangSmith automatically.
# Ensure that the LANGSMITH_TRACING environment variables is set for the wrapper to work.
from langsmith import wrappers
wrapped_client = wrappers.wrap_openai(client)
wrapped_client.chat.completions.create(
  model="gpt-4o-mini",
  messages=messages,
  langsmith_extra={"project_name": "My Project"},
)

# Alternatively, create a RunTree object
# You can set the project name using the project_name parameter
rt = RunTree(
  run_type="llm",
  name="OpenAI Call RunTree",
  inputs={"messages": messages},
  project_name="My Project"
)
chat_completion = client.chat.completions.create(
  model="gpt-4o-mini",
  messages=messages,
)
# End and submit the run
rt.end(outputs=chat_completion)
rt.post()

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

# API key with access to multiple workspaces
api_key = os.getenv("LS_CROSS_WORKSPACE_KEY")

# Initialize clients for different workspaces
workspace_a_client = Client(
    api_key=api_key,
    api_url="https://api.smith.langchain.com",
    workspace_id="<YOUR_WORKSPACE_A_ID>"  # e.g., "abc123..."
)

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

# Example: Route based on customer ID
def get_workspace_client(customer_id: str):
    """Route to appropriate workspace based on customer."""
    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):
    """Process a customer request with workspace-specific tracing."""
    # Your business logic here
    return {"status": "success", "data": data}

# Use tracing_context to route to the appropriate workspace
def handle_customer_request(customer_id: str, request_data: dict):
    client, project_name = get_workspace_client(customer_id)

    # Everything within this context will be traced to the selected workspace
    with tracing_context(enabled=True, client=client, project_name=project_name):
        result = process_request(request_data, customer_id)

    return result

# Example usage
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 key with access to multiple workspaces
api_key = os.getenv("LS_CROSS_WORKSPACE_KEY")

# Initialize clients for different workspaces
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>"
)

# Define configuration schema for workspace routing
class Configuration(TypedDict):
    workspace_id: str

# Define the graph state
class State(TypedDict):
    response: str

def greeting(state: State, config: RunnableConfig) -> State:
    """Generate a workspace-specific greeting."""
    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}

# Build the base graph
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):
    """Dynamically route traces to different workspaces based on configuration."""
    # Extract workspace_id from the configuration
    workspace_id = config.get("configurable", {}).get("workspace_id", "workspace_a")

    # Route to the appropriate workspace
    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"

    # Apply the tracing context for the selected workspace
    with tracing_context(enabled=True, client=client, project_name=project_name):
        yield base_graph

# Usage: Invoke with different workspace configurations
# await graph({"configurable": {"workspace_id": "workspace_a"}})
# await graph({"configurable": {"workspace_id": "workspace_b"}})