Run Management

class RunTree(BaseModel):
    """
    Run schema with back-references for posting runs.

    Represents a single run/span in a trace tree with inputs, outputs,
    timing, metadata, and parent-child relationships.
    """

    # Core identity fields
    name: str
    id: UUID
    run_type: str = "chain"

    # Timing fields
    start_time: datetime
    end_time: Optional[datetime] = None

    # Data fields
    inputs: dict
    outputs: Optional[dict] = None
    error: Optional[str] = None

    # Metadata fields
    extra: dict = Field(default_factory=dict)
    tags: Optional[list[str]] = None
    metadata: Optional[dict] = None

    # Tree structure fields
    parent_run_id: Optional[UUID] = None
    session_name: str  # Project name
    session_id: Optional[UUID] = None  # Project ID
    trace_id: UUID
    dotted_order: str

    # Events
    events: list[dict] = Field(default_factory=list)

    # Reference fields
    reference_example_id: Optional[UUID] = None

    # Serialization fields
    serialized: Optional[dict] = None

    # Child runs (not posted by default)
    child_runs: list["RunTree"] = Field(default_factory=list)

    # Client reference
    client: Optional[Client] = Field(default=None, exclude=True)

def end(
    self,
    *,
    outputs: Optional[dict] = None,
    error: Optional[str] = None,
    end_time: Optional[datetime] = None,
    events: Optional[Sequence[dict]] = None,
) -> None:
    """
    End the run with outputs or error.

    Sets the end_time and optionally updates outputs, error, or events.

    Parameters:
    - outputs: Output data from the run
    - error: Error message if the run failed
    - end_time: When the run ended (defaults to now)
    - events: Additional events to log
    """

def post(
    self,
    *,
    exclude_child_runs: bool = False,
) -> None:
    """
    Post the run to LangSmith.

    Submits the run data to the LangSmith API. Usually called automatically
    when using traceable decorator or trace context manager.

    Parameters:
    - exclude_child_runs: Whether to exclude child runs from the submission
    """

def patch(self) -> None:
    """
    Patch/update the run in LangSmith.

    Updates an existing run with new data.
    """

def wait(self) -> None:
    """
    Wait for the run to be posted.

    Blocks until the run has been successfully submitted to LangSmith.
    """

def create_child(
    self,
    name: str,
    run_type: str = "chain",
    *,
    inputs: Optional[dict] = None,
    outputs: Optional[dict] = None,
    error: Optional[str] = None,
    reference_example_id: Optional[UUID] = None,
    tags: Optional[list[str]] = None,
    extra: Optional[dict] = None,
    metadata: Optional[dict] = None,
    run_id: Optional[UUID] = None,
) -> "RunTree":
    """
    Create a child run.

    Creates a new RunTree as a child of this run, automatically setting
    parent relationships and trace context.

    Parameters:
    - name: Name of the child run
    - run_type: Type of run
    - inputs: Input data
    - outputs: Output data
    - error: Error message
    - reference_example_id: Dataset example ID
    - tags: List of tags
    - extra: Extra metadata
    - metadata: Metadata dictionary
    - run_id: Preset run ID

    Returns:
    New RunTree instance
    """

def add_tags(
    self,
    *tags: str
) -> None:
    """
    Add tags to the run.

    Parameters:
    - *tags: Tag strings to add
    """

def add_metadata(
    self,
    metadata: dict
) -> None:
    """
    Add metadata to the run.

    Merges the provided metadata with existing metadata.

    Parameters:
    - metadata: Dictionary of metadata to add
    """

def add_event(
    self,
    events: Union[dict, Sequence[dict]]
) -> None:
    """
    Add events to the run.

    Events are logged occurrences during run execution.

    Parameters:
    - events: Single event dict or list of event dicts
    """

def add_outputs(
    self,
    outputs: dict
) -> None:
    """
    Add outputs to the run.

    Merges the provided outputs with existing outputs.

    Parameters:
    - outputs: Dictionary of outputs to add
    """

def add_attachment(
    self,
    key: str,
    *,
    mime_type: str,
    data: Optional[bytes] = None,
    path: Optional[Union[str, Path]] = None,
) -> None:
    """
    Add a file attachment to the run.

    Parameters:
    - key: Key for the attachment
    - mime_type: MIME type (e.g., "image/png", "application/pdf")
    - data: Binary data of the attachment
    - path: Path to file to attach (alternative to data)
    """

from langsmith import RunTree, Client
from datetime import datetime

client = Client()

# Create a root run
run = RunTree(
    name="My Operation",
    run_type="chain",
    inputs={"query": "test input"},
    client=client,
    project_name="my-project"
)

# Do some work
result = process()

# End the run
run.end(outputs={"result": result})

# Post to LangSmith
run.post()

from langsmith import RunTree, Client

client = Client()

# Parent run
parent = RunTree(
    name="Parent Operation",
    run_type="chain",
    inputs={"input": "data"},
    client=client
)

# Create child runs
child1 = parent.create_child(
    name="Child 1",
    run_type="tool",
    inputs={"data": "child1 input"}
)
child1.end(outputs={"result": "child1 result"})

child2 = parent.create_child(
    name="Child 2",
    run_type="tool",
    inputs={"data": "child2 input"}
)
child2.end(outputs={"result": "child2 result"})

# End parent
parent.end(outputs={"final": "result"})

# Post the tree (includes children)
parent.post()

from langsmith import RunTree, Client

client = Client()
run = RunTree(
    name="My Run",
    run_type="chain",
    inputs={"input": "data"},
    client=client
)

# Add tags
run.add_tags("production", "v1.0", "important")

# Add metadata
run.add_metadata({
    "user_id": "123",
    "region": "us-west",
    "model": "gpt-4"
})

# Add more metadata later
run.add_metadata({"tokens": 150})

# Add events
run.add_event({"event": "cache_miss", "timestamp": datetime.now()})
run.add_event([
    {"event": "api_call_started"},
    {"event": "api_call_completed"}
])

result = process()
run.end(outputs={"result": result})
run.post()

from langsmith import RunTree, Client

client = Client()
run = RunTree(
    name="Risky Operation",
    run_type="tool",
    inputs={"input": "data"},
    client=client
)

try:
    result = risky_operation()
    run.end(outputs={"result": result})
except Exception as e:
    run.end(error=str(e))
    raise
finally:
    run.post()

from langsmith import RunTree, Client

client = Client()
run = RunTree(
    name="Image Processing",
    run_type="tool",
    inputs={"image_url": "http://example.com/image.png"},
    client=client
)

# Add image as attachment
run.add_attachment(
    "input_image",
    mime_type="image/png",
    path="./input.png"
)

result_image = process_image()

# Add result image
run.add_attachment(
    "output_image",
    mime_type="image/png",
    data=result_image
)

run.end(outputs={"status": "processed"})
run.post()

def configure(
    client: Optional[Client] = ...,
    enabled: Optional[bool] = ...,
    project_name: Optional[str] = ...,
    tags: Optional[list[str]] = ...,
    metadata: Optional[dict[str, Any]] = ...,
) -> None:
    """
    Configure global LangSmith tracing context.

    Call once at application startup to set global defaults for tracing.
    These settings apply to all traced operations unless overridden locally.

    Parameters:
    - client: LangSmith Client instance to use globally (pass None to clear)
    - enabled: Enable/disable tracing globally (True/False/'local'/None)
      - True: Full tracing enabled
      - False: Tracing disabled
      - 'local': Trace locally without sending to server
      - None: Clear the setting
    - project_name: Default project name for all traces (pass None to clear)
    - tags: Global tags to apply to all runs (pass None to clear)
    - metadata: Global metadata to apply to all runs (pass None to clear)
    """

import langsmith as ls
from langsmith import Client

# Basic configuration
ls.configure(
    enabled=True,
    project_name="my-project"
)

# With custom client
custom_client = Client(
    api_key="custom-key",
    api_url="https://custom.langsmith.com"
)
ls.configure(client=custom_client)

# With tags and metadata
ls.configure(
    project_name="production-app",
    tags=["production", "v2.0"],
    metadata={
        "environment": "prod",
        "region": "us-west",
        "version": "2.0.1"
    }
)

# Disable tracing globally
ls.configure(enabled=False)

# Enable local-only tracing (no network calls)
ls.configure(enabled="local")

# Clear specific settings
ls.configure(
    project_name=None,  # Clear project name
    tags=None,  # Clear tags
    metadata=None  # Clear metadata
)

# Multiple configuration calls (later calls override earlier ones)
ls.configure(enabled=True)
ls.configure(project_name="my-project")
ls.configure(tags=["v1"])

# Configuration in different environments
import os

if os.getenv("ENV") == "production":
    ls.configure(
        enabled=True,
        project_name="prod-app",
        tags=["production"]
    )
elif os.getenv("ENV") == "development":
    ls.configure(
        enabled="local",  # Local only in dev
        project_name="dev-app"
    )
else:
    ls.configure(enabled=False)  # Disabled in test

from langsmith import RunTree, Client

client = Client()

# Build a complex trace tree manually
root = RunTree(
    name="Complex Pipeline",
    run_type="chain",
    inputs={"request": "data"},
    client=client
)

# Step 1: Parse input
parse = root.create_child(
    name="Parse Input",
    run_type="parser",
    inputs={"raw": "data"}
)
parsed = parse_input()
parse.end(outputs={"parsed": parsed})

# Step 2: Parallel processing
parallel_parent = root.create_child(
    name="Parallel Processing",
    run_type="chain",
    inputs={"data": parsed}
)

process1 = parallel_parent.create_child(
    name="Process 1",
    run_type="tool",
    inputs={"data": parsed["part1"]}
)
result1 = do_process1()
process1.end(outputs={"result": result1})

process2 = parallel_parent.create_child(
    name="Process 2",
    run_type="tool",
    inputs={"data": parsed["part2"]}
)
result2 = do_process2()
process2.end(outputs={"result": result2})

parallel_parent.end(outputs={"results": [result1, result2]})

# Step 3: Combine results
combine = root.create_child(
    name="Combine Results",
    run_type="chain",
    inputs={"results": [result1, result2]}
)
final = combine_results(result1, result2)
combine.end(outputs={"final": final})

# End root
root.end(outputs={"result": final})
root.post()

from langsmith import traceable, get_current_run_tree

@traceable
def my_function():
    # Get the current RunTree
    run = get_current_run_tree()

    if run:
        # Access run properties
        print(f"Run ID: {run.id}")
        print(f"Run name: {run.name}")
        print(f"Trace ID: {run.trace_id}")
        print(f"Project: {run.session_name}")

        # Modify the run
        run.add_tags("dynamic-tag")
        run.add_metadata({"computed_value": 42})

        # Create a manual child
        child = run.create_child(
            name="Manual Child",
            run_type="tool",
            inputs={"data": "test"}
        )
        child_result = do_child_work()
        child.end(outputs={"result": child_result})
        child.post()

    return "result"

import langsmith as ls
import os

def setup_tracing():
    """Setup tracing based on environment."""
    env = os.getenv("ENVIRONMENT", "development")

    if env == "production":
        ls.configure(
            enabled=True,
            project_name="prod-app",
            tags=["production", "v1.0"],
            metadata={
                "environment": "production",
                "region": os.getenv("REGION", "us-west")
            }
        )
    elif env == "staging":
        ls.configure(
            enabled=True,
            project_name="staging-app",
            tags=["staging"]
        )
    elif env == "development":
        # Local tracing in dev
        ls.configure(
            enabled="local",
            project_name="dev-app"
        )
    else:
        # Disabled for tests
        ls.configure(enabled=False)

# Call at startup
setup_tracing()

from langsmith import RunTree, Client
import time

client = Client()

# Create run
run = RunTree(
    name="Long Operation",
    run_type="chain",
    inputs={"input": "data"},
    client=client
)

# Do work in phases
phase1_result = do_phase1()
run.add_outputs({"phase1": phase1_result})
run.add_event({"event": "phase1_complete"})

time.sleep(1)

phase2_result = do_phase2()
run.add_outputs({"phase2": phase2_result})
run.add_event({"event": "phase2_complete"})

time.sleep(1)

# Final outputs
run.end(outputs={"final_result": combine(phase1_result, phase2_result)})

# Post asynchronously in background
run.post()

# Or wait for posting to complete
run.wait()