import openaifrom langsmith.run_trees import RunTree# This can be a user input to your appquestion = "Can you summarize this morning's meetings?"# Create a top-level runpipeline = RunTree( name="Chat Pipeline", run_type="chain", inputs={"question": question})pipeline.post()# This can be retrieved in a retrieval stepcontext = "During this morning's meeting, we solved all world conflict."messages = [ { "role": "system", "content": "You are a helpful assistant. Please respond to the user's request only based on the given context." }, { "role": "user", "content": f"Question: {question}\nContext: {context}"}]# Create a child runchild_llm_run = pipeline.create_child( name="OpenAI Call", run_type="llm", inputs={"messages": messages},)child_llm_run.post()# Generate a completionclient = openai.Client()chat_completion = client.chat.completions.create( model="gpt-5.4-mini", messages=messages)# End the runs and log themchild_llm_run.end(outputs=chat_completion)child_llm_run.patch()pipeline.end(outputs={"answer": chat_completion.choices[0].message.content})pipeline.patch()
from typing import Any, Callable, Type, TypeVarT = TypeVar("T")def traceable_cls(cls: Type[T]) -> Type[T]: """Instrument all public methods in a class.""" def wrap_method(name: str, method: Any) -> Any: if callable(method) and not name.startswith("__"): return traceable(name=f"{cls.__name__}.{name}")(method) return method # Handle __dict__ case for name in dir(cls): if not name.startswith("_"): try: method = getattr(cls, name) setattr(cls, name, wrap_method(name, method)) except AttributeError: # Skip attributes that can't be set (e.g., some descriptors) pass # Handle __slots__ case if hasattr(cls, "__slots__"): for slot in cls.__slots__: # type: ignore[attr-defined] if not slot.startswith("__"): try: method = getattr(cls, slot) setattr(cls, slot, wrap_method(slot, method)) except AttributeError: # Skip slots that don't have a value yet pass return cls@traceable_clsclass MyClass: def __init__(self, some_val: int): self.some_val = some_val def combine(self, other_val: int): return self.some_val + other_val# See trace: https://smith.langchain.com/public/882f9ecf-5057-426a-ae98-0edf84fdcaf9/rMyClass(13).combine(29)
from langsmith import traceable, uuid7@traceabledef my_pipeline(question: str) -> str: return "answer"run_id = uuid7()my_pipeline("What is the capital of France?", langsmith_extra={"run_id": run_id})# run_id can now be used to attach feedback, query the run, etc.
from langsmith import trace, uuid7run_id = uuid7()with trace("my-pipeline", run_id=run_id) as run: result = "answer" run.end(outputs={"result": result})# run_id can now be used to attach feedback, query the run, etc.