codekingpro/portable-devtools
114k
1"""Schemas for the LangSmith API."""2 3from __future__ import annotations4 5import contextvars6import functools7import json8import logging9import sys10import threading11import urllib.parse12from collections.abc import Mapping, Sequence13from datetime import datetime, timezone14from typing import Any, NamedTuple, Optional, Union, cast15from uuid import UUID16 17from pydantic import ConfigDict, Field, model_validator18from typing_extensions import NotRequired, TypedDict19 20import langsmith._internal._context as _context21from langsmith import schemas as ls_schemas22from langsmith import utils23from langsmith._internal._uuid import uuid7, uuid7_deterministic24from langsmith.client import ID_TYPE, RUN_TYPE_T, Client, _dumps_json, _ensure_uuid25from langsmith.uuid import uuid7_from_datetime26 27logger = logging.getLogger(__name__)28 29 30class ApiKeyAuth(TypedDict):31 """API key authentication for write replicas."""32 33 api_key: str34 35 36class ServiceAuth(TypedDict, total=False):37 """Service-to-service JWT authentication for write replicas."""38 39 service_key: str40 tenant_id: NotRequired[str]41 42 43class AuthHeaders(TypedDict, total=False):44 """Custom authentication headers for write replicas."""45 46 api_key: str47 service_key: str48 tenant_id: str49 authorization: str50 cookie: str51 52 53class WriteReplica(TypedDict, total=False):54 """Configuration for a write replica endpoint."""55 56 api_url: Optional[str]57 api_key: NotRequired[str]58 auth: AuthHeaders59 project_name: Optional[str]60 updates: Optional[dict]61 client: Optional[Client]62 """Optional dedicated :class:`~langsmith.Client` for this replica.63 64 When set, the replica's runs are enqueued on this client's tracing queue65 (and dispatched by its background thread) instead of the RunTree's default66 client. This lets each replica use a different tracing mode — for example,67 one replica with ``Client(tracing_mode="otel")`` and another with the68 default LangSmith-only client.69 70 The field is **not** propagated in distributed-tracing baggage (each service71 must construct its own clients).72 """73 74 75_HEADER_SAFE_REPLICA_FIELDS: frozenset[str] = frozenset({"project_name", "updates"})76 77 78def _filter_replica_for_headers(replica: WriteReplica) -> WriteReplica:79 return cast(80 WriteReplica,81 {k: v for k, v in replica.items() if k in _HEADER_SAFE_REPLICA_FIELDS},82 )83 84 85LANGSMITH_PREFIX = "langsmith-"86LANGSMITH_DOTTED_ORDER = sys.intern(f"{LANGSMITH_PREFIX}trace")87LANGSMITH_DOTTED_ORDER_BYTES = LANGSMITH_DOTTED_ORDER.encode("utf-8")88LANGSMITH_METADATA = sys.intern(f"{LANGSMITH_PREFIX}metadata")89LANGSMITH_TAGS = sys.intern(f"{LANGSMITH_PREFIX}tags")90LANGSMITH_PROJECT = sys.intern(f"{LANGSMITH_PREFIX}project")91LANGSMITH_REPLICAS = sys.intern(f"{LANGSMITH_PREFIX}replicas")92OVERRIDE_OUTPUTS = sys.intern("__omit_auto_outputs")93NOT_PROVIDED = cast(None, object())94_LOCK = threading.Lock()95 96# Context variables97_REPLICAS = contextvars.ContextVar[Optional[Sequence[WriteReplica]]](98 "_REPLICAS", default=None99)100 101_DISTRIBUTED_PARENT_ID = contextvars.ContextVar[Optional[str]](102 "_DISTRIBUTED_PARENT_ID", default=None103)104 105_SENTINEL = cast(None, object())106 107 108def _coerce_to_dict(value):109 if isinstance(value, dict):110 return value111 if (112 not isinstance(value, type)113 and hasattr(value, "model_dump")114 and callable(value.model_dump)115 ):116 return value.model_dump()117 return dict(value)118 119 120TIMESTAMP_LENGTH = 36121 122 123# Note, this is called directly by langchain. Do not remove.124def get_cached_client(**init_kwargs: Any) -> Client:125 global _CLIENT126 if _CLIENT is None:127 with _LOCK:128 if _CLIENT is None:129 _CLIENT = Client(**init_kwargs)130 return _CLIENT131 132 133def configure(134 client: Optional[Client] = _SENTINEL,135 enabled: Optional[bool] = _SENTINEL,136 project_name: Optional[str] = _SENTINEL,137 tags: Optional[list[str]] = _SENTINEL,138 metadata: Optional[dict[str, Any]] = _SENTINEL,139):140 """Configure global LangSmith tracing context.141 142 This function allows you to set global configuration options for LangSmith143 tracing that will be applied to all subsequent traced operations. It modifies144 context variables that control tracing behavior across your application.145 146 Do this once at startup to configure the global settings in code.147 148 If, instead, you wish to only configure tracing for a single invocation,149 use the `tracing_context` context manager instead.150 151 Args:152 client: A LangSmith Client instance to use for all tracing operations.153 154 If provided, this client will be used instead of creating new clients.155 156 Pass `None` to explicitly clear the global client.157 enabled: Whether tracing is enabled.158 159 Can be:160 161 - `True`: Enable tracing and send data to LangSmith162 - `False`: Disable tracing completely163 - `'local'`: Enable tracing but only store data locally164 - `None`: Clear the setting (falls back to environment variables)165 project_name: The LangSmith project name where traces will be sent.166 167 This determines which project dashboard will display your traces.168 169 Pass `None` to explicitly clear the project name.170 tags: A list of tags to be applied to all traced runs.171 172 Tags are useful for filtering and organizing runs in the LangSmith UI.173 174 Pass `None` to explicitly clear all global tags.175 metadata: A dictionary of metadata to attach to all traced runs.176 177 Metadata can store any additional context about your runs.178 179 Pass `None` to explicitly clear all global metadata.180 181 Examples:182 Basic configuration:183 >>> import langsmith as ls184 >>> # Enable tracing with a specific project185 >>> ls.configure(enabled=True, project_name="my-project")186 187 Set global trace masking:188 >>> def hide_keys(data):189 ... if not data:190 ... return {}191 ... return {k: v for k, v in data.items() if k not in ["key1", "key2"]}192 >>> ls.configure(193 ... client=ls.Client(194 ... hide_inputs=hide_keys,195 ... hide_outputs=hide_keys,196 ... )197 ... )198 199 Adding global tags and metadata:200 >>> ls.configure(201 ... tags=["production", "v1.0"],202 ... metadata={"environment": "prod", "version": "1.0.0"},203 ... )204 205 Disabling tracing:206 >>> ls.configure(enabled=False)207 """208 global _CLIENT209 with _LOCK:210 if client is not _SENTINEL:211 _CLIENT = client212 if enabled is not _SENTINEL:213 _context._TRACING_ENABLED.set(enabled)214 _context._GLOBAL_TRACING_ENABLED = enabled215 if project_name is not _SENTINEL:216 _context._PROJECT_NAME.set(project_name)217 _context._GLOBAL_PROJECT_NAME = project_name218 if tags is not _SENTINEL:219 _context._TAGS.set(tags)220 _context._GLOBAL_TAGS = tags221 if metadata is not _SENTINEL:222 _context._METADATA.set(metadata)223 _context._GLOBAL_METADATA = metadata224 225 226def validate_extracted_usage_metadata(227 data: ls_schemas.ExtractedUsageMetadata,228) -> ls_schemas.ExtractedUsageMetadata:229 """Validate that the dict only contains allowed keys."""230 allowed_keys = {231 "input_tokens",232 "output_tokens",233 "total_tokens",234 "input_token_details",235 "output_token_details",236 "input_cost",237 "output_cost",238 "total_cost",239 "input_cost_details",240 "output_cost_details",241 }242 243 extra_keys = set(data.keys()) - allowed_keys244 if extra_keys:245 raise ValueError(f"Unexpected keys in usage metadata: {extra_keys}")246 return data # type: ignore247 248 249class RunTree(ls_schemas.RunBase):250 """Run Schema with back-references for posting runs."""251 252 name: str253 id: UUID = Field(default_factory=uuid7)254 run_type: str = Field(default="chain")255 start_time: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))256 # Note: no longer set.257 parent_run: Optional[RunTree] = Field(default=None, exclude=True)258 parent_dotted_order: Optional[str] = Field(default=None, exclude=True)259 child_runs: list[RunTree] = Field(260 default_factory=list,261 exclude=True,262 )263 session_name: str = Field(264 default_factory=lambda: utils.get_tracer_project() or "default",265 alias="project_name",266 )267 session_id: Optional[UUID] = Field(default=None, alias="project_id")268 extra: dict = Field(default_factory=dict)269 tags: Optional[list[str]] = Field(default_factory=list)270 events: list[dict] = Field(default_factory=list)271 """List of events associated with the run, like272 start and end events."""273 ls_client: Optional[Any] = Field(default=None, exclude=True)274 dotted_order: str = Field(275 default="", description="The order of the run in the tree."276 )277 trace_id: UUID = Field(default="", description="The trace id of the run.") # type: ignore278 dangerously_allow_filesystem: Optional[bool] = Field(279 default=False, description="Whether to allow filesystem access for attachments."280 )281 replicas: Optional[Sequence[WriteReplica]] = Field(282 default=None,283 description="Projects to replicate this run to with optional updates.",284 )285 286 model_config = ConfigDict(287 arbitrary_types_allowed=True,288 populate_by_name=True,289 extra="ignore",290 )291 292 @model_validator(mode="before")293 def infer_defaults(cls, values: dict[str, Any]) -> dict[str, Any]:294 """Assign name to the run."""295 if values.get("name") is None and values.get("serialized") is not None:296 if "name" in values["serialized"]:297 values["name"] = values["serialized"]["name"]298 elif "id" in values["serialized"]:299 values["name"] = values["serialized"]["id"][-1]300 if values.get("name") is None:301 values["name"] = "Unnamed"302 if "client" in values: # Handle user-constructed clients303 values["ls_client"] = values.pop("client")304 elif "_client" in values:305 values["ls_client"] = values.pop("_client")306 if not values.get("ls_client"):307 values["ls_client"] = None308 parent_run = values.pop("parent_run", None)309 if parent_run is not None:310 values["parent_run_id"] = parent_run.id311 values["parent_dotted_order"] = parent_run.dotted_order312 if "id" not in values:313 # Generate UUID from start_time if available314 if "start_time" in values and values["start_time"] is not None:315 values["id"] = uuid7_from_datetime(values["start_time"])316 else:317 now = datetime.now(timezone.utc)318 values["start_time"] = now319 values["id"] = uuid7_from_datetime(now)320 if "trace_id" not in values:321 if parent_run is not None:322 values["trace_id"] = parent_run.trace_id323 else:324 values["trace_id"] = values["id"]325 cast(dict, values.setdefault("extra", {}))326 if values.get("events") is None:327 values["events"] = []328 if values.get("tags") is None:329 values["tags"] = []330 if values.get("outputs") is None:331 values["outputs"] = {}332 for _key in ("inputs", "outputs"):333 _val = values.get(_key)334 if _val is not None and not isinstance(_val, dict):335 values[_key] = _coerce_to_dict(_val)336 if values.get("attachments") is None:337 values["attachments"] = {}338 if values.get("replicas") is None:339 values["replicas"] = _REPLICAS.get()340 values["replicas"] = _ensure_write_replicas(values["replicas"])341 return values342 343 @model_validator(mode="after")344 def ensure_dotted_order(self) -> RunTree:345 """Ensure the dotted order of the run."""346 current_dotted_order = self.dotted_order347 if current_dotted_order and current_dotted_order.strip():348 return self349 current_dotted_order = _create_current_dotted_order(self.start_time, self.id)350 parent_dotted_order = self.parent_dotted_order351 if parent_dotted_order is not None:352 self.dotted_order = parent_dotted_order + "." + current_dotted_order353 else:354 self.dotted_order = current_dotted_order355 return self356 357 @property358 def client(self) -> Client:359 """Return the client."""360 # Lazily load the client361 # If you never use this for API calls, it will never be loaded362 if self.ls_client is None:363 self.ls_client = get_cached_client()364 return self.ls_client365 366 @property367 def _client(self) -> Optional[Client]:368 # For backwards compat369 return self.ls_client370 371 @functools.cached_property372 def trace_start_time(self) -> datetime:373 """Return the start time of the trace (root run)."""374 dt = _parse_dotted_order(self.dotted_order)[0][0]375 return dt.replace(tzinfo=timezone.utc)376 377 def __setattr__(self, name, value):378 """Set the `_client` specially."""379 # For backwards compat380 if name == "_client":381 self.ls_client = value382 else:383 return super().__setattr__(name, value)384 385 def set(386 self,387 *,388 inputs: Optional[Mapping[str, Any]] = NOT_PROVIDED,389 outputs: Optional[Mapping[str, Any]] = NOT_PROVIDED,390 tags: Optional[Sequence[str]] = NOT_PROVIDED,391 metadata: Optional[Mapping[str, Any]] = NOT_PROVIDED,392 usage_metadata: Optional[ls_schemas.ExtractedUsageMetadata] = NOT_PROVIDED,393 ) -> None:394 """Set the inputs, outputs, tags, and metadata of the run.395 396 If performed, this will override the default behavior of the397 end() method to ignore new outputs (that would otherwise be added)398 by the @traceable decorator.399 400 If your LangChain or LangGraph versions are sufficiently up-to-date,401 this will also override the default behavior of `LangChainTracer`.402 403 Args:404 inputs: The inputs to set.405 outputs: The outputs to set.406 tags: The tags to set.407 metadata: The metadata to set.408 usage_metadata: Usage information to set.409 410 Returns:411 None412 """413 if tags is not NOT_PROVIDED:414 self.tags = list(tags)415 if metadata is not NOT_PROVIDED:416 self.extra.setdefault("metadata", {}).update(metadata or {})417 if inputs is not NOT_PROVIDED:418 # Used by LangChain core to determine whether to419 # re-upload the inputs upon run completion420 self.extra["inputs_is_truthy"] = False421 if inputs is None:422 self.inputs = {}423 else:424 self.inputs = _coerce_to_dict(inputs)425 if outputs is not NOT_PROVIDED:426 self.extra[OVERRIDE_OUTPUTS] = True427 if outputs is None:428 self.outputs = {}429 else:430 self.outputs = _coerce_to_dict(outputs)431 if usage_metadata is not NOT_PROVIDED:432 self.extra.setdefault("metadata", {})["usage_metadata"] = (433 validate_extracted_usage_metadata(usage_metadata)434 )435 436 def add_tags(self, tags: Union[Sequence[str], str]) -> None:437 """Add tags to the run."""438 if isinstance(tags, str):439 tags = [tags]440 if self.tags is None:441 self.tags = []442 self.tags.extend(tags)443 444 def add_metadata(self, metadata: dict[str, Any]) -> None:445 """Add metadata to the run."""446 if self.extra is None:447 self.extra = {}448 metadata_: dict = cast(dict, self.extra).setdefault("metadata", {})449 metadata_.update(metadata)450 451 def add_outputs(self, outputs: dict[str, Any]) -> None:452 """Upsert the given outputs into the run.453 454 Args:455 outputs: A dictionary containing the outputs to be added.456 """457 if self.outputs is None:458 self.outputs = {}459 self.outputs.update(outputs)460 461 def add_inputs(self, inputs: dict[str, Any]) -> None:462 """Upsert the given inputs into the run.463 464 Args:465 inputs: A dictionary containing the inputs to be added.466 """467 if self.inputs is None:468 self.inputs = {}469 self.inputs.update(inputs)470 # Set to False so LangChain things it needs to471 # re-upload inputs472 self.extra["inputs_is_truthy"] = False473 474 def add_event(475 self,476 events: Union[477 ls_schemas.RunEvent,478 Sequence[ls_schemas.RunEvent],479 Sequence[dict],480 dict,481 str,482 ],483 ) -> None:484 """Add an event to the list of events.485 486 Args:487 events: The event(s) to be added. It can be a single event, a sequence488 of events, a sequence of dictionaries, a dictionary, or a string.489 490 Returns:491 None492 """493 if self.events is None:494 self.events = []495 if isinstance(events, dict):496 self.events.append(events) # type: ignore[arg-type]497 elif isinstance(events, str):498 self.events.append(499 {500 "name": "event",501 "time": datetime.now(timezone.utc).isoformat(),502 "message": events,503 }504 )505 else:506 self.events.extend(events) # type: ignore[arg-type]507 508 def end(509 self,510 *,511 outputs: Optional[dict] = None,512 error: Optional[str] = None,513 end_time: Optional[datetime] = None,514 events: Optional[Sequence[ls_schemas.RunEvent]] = None,515 metadata: Optional[dict[str, Any]] = None,516 ) -> None:517 """Set the end time of the run and all child runs."""518 self.end_time = end_time or datetime.now(timezone.utc)519 # We've already 'set' the outputs, so ignore520 # the ones that are automatically included521 if not self.extra.get(OVERRIDE_OUTPUTS):522 if outputs is not None:523 dict_outputs = _coerce_to_dict(outputs)524 if not self.outputs:525 self.outputs = dict_outputs526 else:527 self.outputs.update(dict_outputs)528 if error is not None:529 self.error = error530 if events is not None:531 self.add_event(events)532 if metadata is not None:533 self.add_metadata(metadata)534 535 def create_child(536 self,537 name: str,538 run_type: RUN_TYPE_T = "chain",539 *,540 run_id: Optional[ID_TYPE] = None,541 serialized: Optional[dict] = None,542 inputs: Optional[dict] = None,543 outputs: Optional[dict] = None,544 error: Optional[str] = None,545 reference_example_id: Optional[UUID] = None,546 start_time: Optional[datetime] = None,547 end_time: Optional[datetime] = None,548 tags: Optional[list[str]] = None,549 extra: Optional[dict] = None,550 attachments: Optional[ls_schemas.Attachments] = None,551 ) -> RunTree:552 """Add a child run to the run tree."""553 # Ensure child start_time is never earlier than parent start_time554 # to prevent timestamp ordering violations in dotted_order555 if start_time is not None and self.start_time is not None:556 if start_time < self.start_time:557 logger.debug(558 f"Adjusting child run '{name}' start_time from {start_time} "559 f"to {self.start_time} to maintain timestamp ordering with "560 f"parent '{self.name}'"561 )562 start_time = max(start_time, self.start_time)563 564 serialized_ = serialized or {"name": name}565 if extra:566 child_extra = dict(extra)567 child_meta = (extra.get("metadata") or {}).copy()568 else:569 child_extra = {}570 child_meta = {}571 parent_meta = (self.extra or {}).get("metadata") or {}572 child_extra["metadata"] = {**parent_meta, **child_meta}573 run = RunTree(574 name=name,575 id=_ensure_uuid(run_id),576 serialized=serialized_,577 inputs=inputs or {},578 outputs=outputs or {},579 error=error,580 run_type=run_type,581 reference_example_id=reference_example_id,582 start_time=start_time or datetime.now(timezone.utc),583 end_time=end_time,584 extra=child_extra,585 parent_run=self,586 project_name=self.session_name,587 replicas=self.replicas,588 ls_client=self.ls_client,589 tags=tags,590 attachments=attachments or {}, # type: ignore591 dangerously_allow_filesystem=self.dangerously_allow_filesystem,592 )593 594 return run595 596 def _get_dicts_safe(self):597 # Things like generators cannot be copied598 self_dict = self.model_dump(599 exclude={"child_runs", "inputs", "outputs"}, exclude_none=True600 )601 if self.inputs is not None:602 # shallow copy. deep copying will occur in the client603 inputs_ = {}604 attachments = self_dict.get("attachments", {})605 for k, v in self.inputs.items():606 if isinstance(v, ls_schemas.Attachment):607 attachments[k] = v608 else:609 inputs_[k] = v610 self_dict["inputs"] = inputs_611 if attachments:612 self_dict["attachments"] = attachments613 if self.outputs is not None:614 # shallow copy; deep copying will occur in the client615 self_dict["outputs"] = self.outputs.copy()616 return self_dict617 618 def _slice_parent_id(self, parent_id: str, run_dict: dict) -> None:619 """Slice the parent id from dotted order.620 621 Additionally check if the current run is a child of the parent. If so, update622 the parent_run_id to None, and set the trace id to the new root id after623 parent_id.624 """625 if dotted_order := run_dict.get("dotted_order"):626 segs = dotted_order.split(".")627 start_idx = None628 parent_id = str(parent_id)629 # TODO(angus): potentially use binary search to find the index630 for idx, part in enumerate(segs):631 seg_id = part[-TIMESTAMP_LENGTH:]632 if str(seg_id) == parent_id:633 start_idx = idx634 break635 if start_idx is not None:636 # Trim segments to start after parent_id (exclusive)637 trimmed_segs = segs[start_idx + 1 :]638 # Rebuild dotted_order639 run_dict["dotted_order"] = ".".join(trimmed_segs)640 if trimmed_segs:641 run_dict["trace_id"] = UUID(trimmed_segs[0][-TIMESTAMP_LENGTH:])642 else:643 run_dict["trace_id"] = run_dict["id"]644 if str(run_dict.get("parent_run_id")) == parent_id:645 # We've found the new root node.646 run_dict.pop("parent_run_id", None)647 648 def _remap_for_project(649 self, project_name: str, updates: Optional[dict] = None650 ) -> dict:651 """Rewrites ids/dotted_order for a given project with optional updates."""652 run_dict = self._get_dicts_safe()653 if project_name == self.session_name:654 return run_dict655 656 if updates and updates.get("reroot", False):657 distributed_parent_id = _DISTRIBUTED_PARENT_ID.get()658 if distributed_parent_id:659 self._slice_parent_id(distributed_parent_id, run_dict)660 661 old_id = run_dict["id"]662 new_id = uuid7_deterministic(UUID(str(old_id)), project_name)663 # trace id664 old_trace = run_dict.get("trace_id")665 if old_trace:666 new_trace = uuid7_deterministic(UUID(str(old_trace)), project_name)667 else:668 new_trace = None669 # parent id670 parent = run_dict.get("parent_run_id")671 if parent:672 new_parent = uuid7_deterministic(UUID(str(parent)), project_name)673 else:674 new_parent = None675 # dotted order676 if run_dict.get("dotted_order"):677 segs = run_dict["dotted_order"].split(".")678 rebuilt = []679 for part in segs[:-1]:680 seg_id = UUID(part[-TIMESTAMP_LENGTH:])681 repl = uuid7_deterministic(seg_id, project_name)682 rebuilt.append(part[:-TIMESTAMP_LENGTH] + str(repl))683 rebuilt.append(segs[-1][:-TIMESTAMP_LENGTH] + str(new_id))684 dotted = ".".join(rebuilt)685 else:686 dotted = None687 dup = utils.deepish_copy(run_dict)688 dup.update(689 {690 "id": new_id,691 "trace_id": new_trace,692 "parent_run_id": new_parent,693 "dotted_order": dotted,694 "session_name": project_name,695 }696 )697 if updates:698 dup.update(updates)699 return dup700 701 def post(self, exclude_child_runs: bool = True) -> None:702 """Post the run tree to the API asynchronously."""703 if self.replicas:704 for replica in self.replicas:705 project_name = replica.get("project_name") or self.session_name706 updates = replica.get("updates")707 run_dict = self._remap_for_project(project_name, updates)708 api_url, api_key, service_key, tenant_id, authorization, cookie = (709 _extract_replica_auth(replica)710 )711 replica_client = replica.get("client") or self.client712 if not hasattr(replica_client, "create_run"):713 raise TypeError(714 f"WriteReplica 'client' must be a langsmith.Client, "715 f"got {type(replica_client).__name__}"716 )717 replica_client.create_run(718 **run_dict,719 api_key=api_key,720 api_url=api_url,721 service_key=service_key,722 tenant_id=tenant_id,723 authorization=authorization,724 cookie=cookie,725 )726 else:727 kwargs = self._get_dicts_safe()728 self.client.create_run(**kwargs)729 if self.attachments:730 keys = [str(name) for name in self.attachments]731 self.events.append(732 {733 "name": "uploaded_attachment",734 "time": datetime.now(timezone.utc).isoformat(),735 "message": set(keys),736 }737 )738 if not exclude_child_runs:739 for child_run in self.child_runs:740 child_run.post(exclude_child_runs=False)741 742 def patch(self, *, exclude_inputs: bool = False) -> None:743 """Patch the run tree to the API in a background thread.744 745 Args:746 exclude_inputs: Whether to exclude inputs from the patch request.747 """748 if not self.end_time:749 self.end()750 attachments = {751 a: v for a, v in self.attachments.items() if isinstance(v, tuple)752 }753 try:754 # Avoid loading the same attachment twice755 if attachments:756 uploaded = next(757 (758 ev759 for ev in self.events760 if ev.get("name") == "uploaded_attachment"761 ),762 None,763 )764 if uploaded:765 attachments = {766 a: v767 for a, v in attachments.items()768 if a not in uploaded["message"]769 }770 except Exception as e:771 logger.warning(f"Error filtering attachments to upload: {e}")772 if self.replicas:773 for replica in self.replicas:774 project_name = replica.get("project_name") or self.session_name775 updates = replica.get("updates")776 run_dict = self._remap_for_project(project_name, updates)777 api_url, api_key, service_key, tenant_id, authorization, cookie = (778 _extract_replica_auth(replica)779 )780 replica_client = replica.get("client") or self.client781 if not hasattr(replica_client, "update_run"):782 raise TypeError(783 f"WriteReplica 'client' must be a langsmith.Client, "784 f"got {type(replica_client).__name__}"785 )786 replica_client.update_run(787 name=run_dict["name"],788 run_id=run_dict["id"],789 run_type=run_dict.get("run_type"),790 start_time=run_dict.get("start_time"),791 inputs=None if exclude_inputs else run_dict["inputs"],792 outputs=run_dict["outputs"],793 error=run_dict.get("error"),794 parent_run_id=run_dict.get("parent_run_id"),795 session_name=run_dict.get("session_name"),796 reference_example_id=run_dict.get("reference_example_id"),797 end_time=run_dict.get("end_time"),798 dotted_order=run_dict.get("dotted_order"),799 trace_id=run_dict.get("trace_id"),800 events=run_dict.get("events"),801 tags=run_dict.get("tags"),802 extra=run_dict.get("extra"),803 attachments=attachments,804 api_key=api_key,805 api_url=api_url,806 service_key=service_key,807 tenant_id=tenant_id,808 authorization=authorization,809 cookie=cookie,810 )811 else:812 self.client.update_run(813 name=self.name,814 run_id=self.id,815 run_type=cast(RUN_TYPE_T, self.run_type),816 start_time=self.start_time,817 inputs=(818 None819 if exclude_inputs820 else (self.inputs.copy() if self.inputs else None)821 ),822 outputs=self.outputs.copy() if self.outputs else None,823 error=self.error,824 parent_run_id=self.parent_run_id,825 session_name=self.session_name,826 reference_example_id=self.reference_example_id,827 end_time=self.end_time,828 dotted_order=self.dotted_order,829 trace_id=self.trace_id,830 events=self.events,831 tags=self.tags,832 extra=self.extra,833 attachments=attachments,834 )835 836 def wait(self) -> None:837 """Wait for all `_futures` to complete."""838 pass839 840 def get_url(self) -> str:841 """Return the URL of the run."""842 return self.client.get_run_url(run=self)843 844 @classmethod845 def from_dotted_order(846 cls,847 dotted_order: str,848 **kwargs: Any,849 ) -> RunTree:850 """Create a new 'child' span from the provided dotted order.851 852 Returns:853 RunTree: The new span.854 """855 headers = {856 LANGSMITH_DOTTED_ORDER: dotted_order,857 }858 return cast(RunTree, cls.from_headers(headers, **kwargs)) # type: ignore[arg-type]859 860 @classmethod861 def from_runnable_config(862 cls,863 config: Optional[dict],864 **kwargs: Any,865 ) -> Optional[RunTree]:866 """Create a new 'child' span from the provided runnable config.867 868 Requires `langchain` to be installed.869 870 Returns:871 The new span or `None` if no parent span information is found.872 """873 try:874 from langchain_core.callbacks.manager import (875 AsyncCallbackManager,876 CallbackManager,877 )878 from langchain_core.runnables import RunnableConfig, ensure_config879 from langchain_core.tracers.langchain import LangChainTracer880 except ImportError as e:881 raise ImportError(882 "RunTree.from_runnable_config requires langchain-core to be installed. "883 "You can install it with `pip install langchain-core`."884 ) from e885 if config is None:886 config_ = ensure_config(887 cast(RunnableConfig, config) if isinstance(config, dict) else None888 )889 else:890 config_ = cast(RunnableConfig, config)891 892 if (893 (cb := config_.get("callbacks"))894 and isinstance(cb, (CallbackManager, AsyncCallbackManager))895 and cb.parent_run_id896 and (897 tracer := next(898 (t for t in cb.handlers if isinstance(t, LangChainTracer)),899 None,900 )901 )902 ):903 if (run := tracer.run_map.get(str(cb.parent_run_id))) and run.dotted_order:904 dotted_order = run.dotted_order905 kwargs["run_type"] = run.run_type906 kwargs["inputs"] = run.inputs907 kwargs["outputs"] = run.outputs908 kwargs["start_time"] = run.start_time909 kwargs["end_time"] = run.end_time910 kwargs["tags"] = sorted(set(run.tags or [] + kwargs.get("tags", [])))911 kwargs["name"] = run.name912 extra_ = kwargs.setdefault("extra", {})913 metadata_ = extra_.setdefault("metadata", {})914 metadata_.update(run.metadata)915 elif hasattr(tracer, "order_map") and cb.parent_run_id in tracer.order_map:916 dotted_order = tracer.order_map[cb.parent_run_id][1]917 else:918 return None919 kwargs["client"] = tracer.client920 kwargs["project_name"] = tracer.project_name921 return RunTree.from_dotted_order(dotted_order, **kwargs)922 return None923 924 @classmethod925 def from_headers(926 cls, headers: Mapping[Union[str, bytes], Union[str, bytes]], **kwargs: Any927 ) -> Optional[RunTree]:928 """Create a new 'parent' span from the provided headers.929 930 Extracts parent span information from the headers and creates a new span.931 932 Metadata and tags are extracted from the baggage header.933 934 The dotted order and trace id are extracted from the trace header.935 936 Returns:937 The new span or `None` if no parent span information is found.938 """939 init_args = kwargs.copy()940 941 langsmith_trace = cast(Optional[str], headers.get(LANGSMITH_DOTTED_ORDER))942 if not langsmith_trace:943 langsmith_trace_bytes = cast(944 Optional[bytes], headers.get(LANGSMITH_DOTTED_ORDER_BYTES)945 )946 if not langsmith_trace_bytes:947 return # type: ignore[return-value]948 langsmith_trace = langsmith_trace_bytes.decode("utf-8")949 950 parent_dotted_order = langsmith_trace.strip()951 parsed_dotted_order = _parse_dotted_order(parent_dotted_order)952 trace_id = parsed_dotted_order[0][1]953 init_args["trace_id"] = trace_id954 init_args["id"] = parsed_dotted_order[-1][1]955 init_args["dotted_order"] = parent_dotted_order956 if len(parsed_dotted_order) >= 2:957 # Has a parent958 init_args["parent_run_id"] = parsed_dotted_order[-2][1]959 # All placeholders. We assume the source process960 # handles the life-cycle of the run.961 init_args["start_time"] = init_args.get("start_time") or datetime.now(962 timezone.utc963 )964 init_args["run_type"] = init_args.get("run_type") or "chain"965 init_args["name"] = init_args.get("name") or "parent"966 967 baggage = _Baggage.from_headers(headers)968 if baggage.metadata or baggage.tags:969 init_args["extra"] = init_args.setdefault("extra", {})970 init_args["extra"]["metadata"] = init_args["extra"].setdefault(971 "metadata", {}972 )973 metadata = {**baggage.metadata, **init_args["extra"]["metadata"]}974 init_args["extra"]["metadata"] = metadata975 tags = sorted(set(baggage.tags + init_args.get("tags", [])))976 init_args["tags"] = tags977 if baggage.project_name:978 init_args["project_name"] = baggage.project_name979 if baggage.replicas:980 init_args["replicas"] = baggage.replicas981 982 run_tree = RunTree(**init_args)983 984 # Set the distributed parent ID to this run's ID for rerooting985 _DISTRIBUTED_PARENT_ID.set(str(run_tree.id))986 987 return run_tree988 989 def to_headers(self) -> dict[str, str]:990 """Return the `RunTree` as a dictionary of headers."""991 headers = {}992 if self.trace_id:993 headers[f"{LANGSMITH_DOTTED_ORDER}"] = self.dotted_order994 baggage = _Baggage(995 metadata=self.extra.get("metadata", {}),996 tags=self.tags,997 project_name=self.session_name,998 replicas=self.replicas,999 )1000 headers["baggage"] = baggage.to_header()1001 return headers1002 1003 def __repr__(self):1004 """Return a string representation of the `RunTree` object."""1005 return (1006 f"RunTree(id={self.id}, name='{self.name}', "1007 f"run_type='{self.run_type}', dotted_order='{self.dotted_order}')"1008 )1009 1010 1011class _Baggage:1012 """Baggage header information."""1013 1014 def __init__(1015 self,1016 metadata: Optional[dict[str, str]] = None,1017 tags: Optional[list[str]] = None,1018 project_name: Optional[str] = None,1019 replicas: Optional[Sequence[WriteReplica]] = None,1020 ):1021 """Initialize the Baggage object."""1022 self.metadata = metadata or {}1023 self.tags = tags or []1024 self.project_name = project_name1025 self.replicas = replicas or []1026 1027 @classmethod1028 def from_header(cls, header_value: Optional[str]) -> _Baggage:1029 """Create a Baggage object from the given header value."""1030 if not header_value:1031 return cls()1032 metadata = {}1033 tags = []1034 project_name = None1035 replicas: Optional[list[WriteReplica]] = None1036 try:1037 for item in header_value.split(","):1038 key, value = item.split("=", 1)1039 if key == LANGSMITH_METADATA:1040 metadata = json.loads(urllib.parse.unquote(value))1041 elif key == LANGSMITH_TAGS:1042 tags = urllib.parse.unquote(value).split(",")1043 elif key == LANGSMITH_PROJECT:1044 project_name = urllib.parse.unquote(value)1045 elif key == LANGSMITH_REPLICAS:1046 replicas_data = json.loads(urllib.parse.unquote(value))1047 parsed_replicas: list[WriteReplica] = []1048 for replica_item in replicas_data:1049 if (1050 isinstance(replica_item, (tuple, list))1051 and len(replica_item) == 21052 ):1053 # Convert legacy format to WriteReplica1054 parsed_replicas.append(1055 WriteReplica(1056 api_url=None,1057 project_name=str(replica_item[0]),1058 updates=replica_item[1],1059 )1060 )1061 elif isinstance(replica_item, dict):1062 filtered_replica = _filter_replica_for_headers(1063 cast(WriteReplica, replica_item)1064 )1065 if filtered_replica.get("project_name"):1066 parsed_replicas.append(filtered_replica)1067 else:1068 logger.warning(1069 f"Unknown replica format in baggage: {replica_item}"1070 )1071 continue1072 replicas = parsed_replicas1073 except Exception as e:1074 logger.warning(f"Error parsing baggage header: {e}")1075 1076 return cls(1077 metadata=metadata, tags=tags, project_name=project_name, replicas=replicas1078 )1079 1080 @classmethod1081 def from_headers(cls, headers: Mapping[Union[str, bytes], Any]) -> _Baggage:1082 if "baggage" in headers:1083 return cls.from_header(headers["baggage"])1084 elif b"baggage" in headers:1085 return cls.from_header(cast(bytes, headers[b"baggage"]).decode("utf-8"))1086 else:1087 return cls.from_header(None)1088 1089 def to_header(self) -> str:1090 """Return the Baggage object as a header value."""1091 items = []1092 if self.metadata:1093 serialized_metadata = _dumps_json(self.metadata)1094 items.append(1095 f"{LANGSMITH_PREFIX}metadata={urllib.parse.quote(serialized_metadata)}"1096 )1097 if self.tags:1098 serialized_tags = ",".join(self.tags)1099 items.append(1100 f"{LANGSMITH_PREFIX}tags={urllib.parse.quote(serialized_tags)}"1101 )1102 if self.project_name:1103 items.append(1104 f"{LANGSMITH_PREFIX}project={urllib.parse.quote(self.project_name)}"1105 )1106 return ",".join(items)1107 1108 1109@functools.lru_cache(maxsize=1)1110def _parse_write_replicas_from_env_var(env_var: Optional[str]) -> list[WriteReplica]:1111 """Parse write replicas from LANGSMITH_RUNS_ENDPOINTS environment variable value.1112 1113 Supports array format [{"api_url": "x", "api_key": "y"}] and object format1114 {"url": "key"}.1115 """1116 if not env_var:1117 return []1118 1119 try:1120 parsed = json.loads(env_var)1121 1122 if isinstance(parsed, list):1123 replicas = []1124 for item in parsed:1125 if not isinstance(item, dict):1126 logger.warning(1127 f"Invalid item type in LANGSMITH_RUNS_ENDPOINTS: "1128 f"expected dict, got {type(item).__name__}"1129 )1130 continue1131 1132 api_url = item.get("api_url")1133 api_key = item.get("api_key")1134 1135 if not isinstance(api_url, str):1136 logger.warning(1137 f"Invalid api_url type in LANGSMITH_RUNS_ENDPOINTS: "1138 f"expected string, got {type(api_url).__name__}"1139 )1140 continue1141 1142 if not isinstance(api_key, str):1143 logger.warning(1144 f"Invalid api_key type in LANGSMITH_RUNS_ENDPOINTS: "1145 f"expected string, got {type(api_key).__name__}"1146 )1147 continue1148 1149 replicas.append(1150 WriteReplica(1151 api_url=api_url.rstrip("/"),1152 auth=AuthHeaders(api_key=api_key),1153 project_name=None,1154 updates=None,1155 )1156 )1157 return replicas1158 elif isinstance(parsed, dict):1159 _check_endpoint_env_unset(parsed)1160 1161 replicas = []1162 for url, key in parsed.items():1163 url = url.rstrip("/")1164 1165 if isinstance(key, str):1166 replicas.append(1167 WriteReplica(1168 api_url=url,1169 auth=AuthHeaders(api_key=key),1170 project_name=None,1171 updates=None,1172 )1173 )1174 else:1175 logger.warning(1176 f"Invalid value type in LANGSMITH_RUNS_ENDPOINTS for URL "1177 f"{url}: "1178 f"expected string, got {type(key).__name__}"1179 )1180 continue1181 return replicas1182 else:1183 logger.warning(1184 f"Invalid LANGSMITH_RUNS_ENDPOINTS – must be valid JSON list of "1185 "objects with api_url and api_key properties, or object mapping "1186 f"url->apiKey, got {type(parsed).__name__}"1187 )1188 return []1189 except utils.LangSmithUserError:1190 raise1191 except Exception as e:1192 logger.warning(1193 "Invalid LANGSMITH_RUNS_ENDPOINTS – must be valid JSON list of "1194 f"objects with api_url and api_key properties, or object mapping"1195 f" url->apiKey: {e}"1196 )1197 return []1198 1199 1200def _get_write_replicas_from_env() -> list[WriteReplica]: