diff --git a/langfuse/__init__.py b/langfuse/__init__.py index 6f2bfbb08..222d01079 100644 --- a/langfuse/__init__.py +++ b/langfuse/__init__.py @@ -65,6 +65,7 @@ from ._client.get_client import get_client from ._client.observe import observe from ._client.propagation import propagate_attributes +from ._client.skills import discover_skills from ._client.span import ( LangfuseAgent, LangfuseChain, @@ -93,6 +94,8 @@ OtelSpanData, OtelSpanIdentifier, OtelSpanPatch, + SkillReference, + SkillsConfig, ) Langfuse = _client_module.Langfuse @@ -104,6 +107,9 @@ "get_client", "observe", "propagate_attributes", + "discover_skills", + "SkillReference", + "SkillsConfig", "ObservationTypeLiteral", "LangfuseSpan", "LangfuseGeneration", diff --git a/langfuse/_client/attributes.py b/langfuse/_client/attributes.py index c3ca70a09..0e8478500 100644 --- a/langfuse/_client/attributes.py +++ b/langfuse/_client/attributes.py @@ -18,11 +18,12 @@ ObservationTypeGenerationLike, ObservationTypeSpanLike, ) +from langfuse._client.skills import serialize_available_skills from langfuse._utils.serializer import EventSerializer from langfuse.api import MapValue from langfuse.logger import langfuse_logger from langfuse.model import PromptClient -from langfuse.types import SpanLevel +from langfuse.types import SkillReference, SpanLevel class LangfuseOtelSpanAttributes: @@ -41,6 +42,7 @@ class LangfuseOtelSpanAttributes: OBSERVATION_STATUS_MESSAGE = "langfuse.observation.status_message" OBSERVATION_INPUT = "langfuse.observation.input" OBSERVATION_OUTPUT = "langfuse.observation.output" + OBSERVATION_SKILLS_AVAILABLE = "langfuse.observation.skills_available" # Langfuse-observation of type Generation attributes OBSERVATION_COMPLETION_START_TIME = "langfuse.observation.completion_start_time" @@ -87,6 +89,7 @@ def create_trace_attributes( def create_span_attributes( *, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, input: Optional[Any] = None, output: Optional[Any] = None, level: Optional[SpanLevel] = None, @@ -103,6 +106,7 @@ def create_span_attributes( LangfuseOtelSpanAttributes.VERSION: version, LangfuseOtelSpanAttributes.OBSERVATION_INPUT: _serialize(input), LangfuseOtelSpanAttributes.OBSERVATION_OUTPUT: _serialize(output), + **create_available_skills_attributes(available_skills), **_flatten_and_serialize_metadata(metadata, "observation"), } @@ -114,6 +118,7 @@ def create_generation_attributes( name: Optional[str] = None, completion_start_time: Optional[datetime] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, version: Optional[str] = None, @@ -148,12 +153,22 @@ def create_generation_attributes( LangfuseOtelSpanAttributes.OBSERVATION_MODEL_PARAMETERS: _serialize( model_parameters ), + **create_available_skills_attributes(available_skills), **_flatten_and_serialize_metadata(metadata, "observation"), } return {k: v for k, v in attributes.items() if v is not None} +def create_available_skills_attributes( + available_skills: Optional[list[SkillReference]], +) -> dict[str, str]: + serialized = serialize_available_skills(available_skills) + if serialized is None: + return {} + return {LangfuseOtelSpanAttributes.OBSERVATION_SKILLS_AVAILABLE: serialized} + + def _serialize(obj: Any) -> Optional[str]: if obj is None or isinstance(obj, str): return obj diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index df04f9d98..e926e4571 100644 --- a/langfuse/_client/client.py +++ b/langfuse/_client/client.py @@ -144,6 +144,8 @@ MaskFunction, MaskOtelSpansFunction, ScoreDataType, + SkillReference, + SkillsConfig, SpanLevel, TraceContext, ) @@ -252,6 +254,7 @@ def mask_otel_spans( additional_headers (Optional[Dict[str, str]]): Additional headers to include in all API requests and in the default OTLPSpanExporter requests. These headers will be merged with default headers. Note: If httpx_client is provided, additional_headers must be set directly on your custom httpx_client as well. If `span_exporter` is provided, these headers are not wired into that exporter and must be configured on the exporter instance directly. tracer_provider(Optional[TracerProvider]): OpenTelemetry TracerProvider to use for Langfuse. This can be useful to set to have disconnected tracing between Langfuse and other OpenTelemetry-span emitting libraries. Note: To track active spans, the context is still shared between TracerProviders. This may lead to broken trace trees. id_generator (Optional[IdGenerator]): OpenTelemetry ID generator to use when Langfuse creates its own TracerProvider. If omitted, the OpenTelemetry SDK default is used. If `tracer_provider` is provided, or an OpenTelemetry TracerProvider is already registered globally, configure the ID generator on that provider instead. + skills (Optional[SkillsConfig]): Configure available skills; attach them with propagate_attributes(attach_skills=True). Shared clients use the first configuration for their public key. span_exporter (Optional[SpanExporter]): Custom OpenTelemetry span exporter for the Langfuse span processor. If omitted, Langfuse creates an OTLPSpanExporter pointed at the Langfuse OTLP endpoint. If provided, Langfuse does not wire `base_url`, exporter headers, exporter auth, exporter timeout, `otel_compression`, or the `LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES` request size limit into it. Configure endpoint, headers, timeout, and compression on the exporter instance directly. If you are sending spans to Langfuse v4 or using Langfuse Cloud Fast Preview, include `x-langfuse-ingestion-version=4` on the exporter to enable real time processing of exported spans. otel_compression (Optional[Literal["gzip", "none"]]): Compression for span batches sent by the default OTLP span exporter: "gzip" (default) or "none" to send uncompressed. If unset, the LANGFUSE_OTEL_COMPRESSION, OTEL_EXPORTER_OTLP_TRACES_COMPRESSION and OTEL_EXPORTER_OTLP_COMPRESSION environment variables apply, in that order, then gzip. Values are case-insensitive; an invalid value falls through to the next setting. @@ -320,6 +323,7 @@ def __init__( id_generator: Optional[IdGenerator] = None, span_exporter: Optional[SpanExporter] = None, otel_compression: Optional[Literal["gzip", "none"]] = None, + skills: Optional[SkillsConfig] = None, ): self._base_url = ( base_url @@ -408,6 +412,7 @@ def __init__( id_generator=id_generator, span_exporter=span_exporter, otel_compression=otel_compression, + skills=skills, ) self._mask = self._resources.mask @@ -486,6 +491,7 @@ def start_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -507,6 +513,7 @@ def start_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -522,6 +529,7 @@ def start_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -537,6 +545,7 @@ def start_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -552,6 +561,7 @@ def start_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -567,6 +577,7 @@ def start_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -582,6 +593,7 @@ def start_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -597,6 +609,7 @@ def start_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -618,6 +631,7 @@ def start_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -632,6 +646,7 @@ def start_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -664,6 +679,7 @@ def start_observation( input: Input data for the operation output: Output data from the operation metadata: Additional metadata to associate with the observation + available_skills: Skills available to this observation. version: Version identifier for the code or component level: Importance level of the observation status_message: Optional status message for the observation @@ -689,7 +705,9 @@ def start_observation( with otel_trace_api.use_span( cast(otel_trace_api.Span, remote_parent_span) ): - otel_span = self._otel_tracer.start_span(name=name) + otel_span = self._otel_tracer.start_span( + name=name, + ) _set_span_attributes_within_limit( otel_span, {LangfuseOtelSpanAttributes.AS_ROOT: True} ) @@ -700,6 +718,7 @@ def start_observation( input=input, output=output, metadata=metadata, + available_skills=available_skills, version=version, level=level, status_message=status_message, @@ -719,6 +738,7 @@ def start_observation( input=input, output=output, metadata=metadata, + available_skills=available_skills, version=version, level=level, status_message=status_message, @@ -738,6 +758,7 @@ def _create_observation_from_otel_span( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -771,6 +792,7 @@ def _create_observation_from_otel_span( input=input, output=output, metadata=metadata, + available_skills=available_skills, version=version, level=level, status_message=status_message, @@ -794,6 +816,7 @@ def _create_observation_from_otel_span( input=input, output=output, metadata=metadata, + available_skills=available_skills, version=version, level=level, status_message=status_message, @@ -812,6 +835,7 @@ def start_as_current_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -834,6 +858,7 @@ def start_as_current_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -850,6 +875,7 @@ def start_as_current_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -866,6 +892,7 @@ def start_as_current_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -882,6 +909,7 @@ def start_as_current_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -898,6 +926,7 @@ def start_as_current_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -914,6 +943,7 @@ def start_as_current_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -930,6 +960,7 @@ def start_as_current_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -952,6 +983,7 @@ def start_as_current_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -967,6 +999,7 @@ def start_as_current_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -1003,6 +1036,7 @@ def start_as_current_observation( input: Input data for the operation (can be any JSON-serializable object) output: Output data from the operation (can be any JSON-serializable object) metadata: Additional metadata to associate with the observation + available_skills: Skills available to this observation. version: Version identifier for the code or component level: Importance level of the observation (info, warning, error) status_message: Optional status message for the observation @@ -1073,6 +1107,7 @@ def start_as_current_observation( input=input, output=output, metadata=metadata, + available_skills=available_skills, version=version, level=level, status_message=status_message, @@ -1097,6 +1132,7 @@ def start_as_current_observation( input=input, output=output, metadata=metadata, + available_skills=available_skills, version=version, level=level, status_message=status_message, @@ -1138,6 +1174,7 @@ def start_as_current_observation( input=input, output=output, metadata=metadata, + available_skills=available_skills, version=version, level=level, status_message=status_message, @@ -1161,6 +1198,7 @@ def start_as_current_observation( input=input, output=output, metadata=metadata, + available_skills=available_skills, version=version, level=level, status_message=status_message, @@ -1178,6 +1216,7 @@ def start_as_current_observation( input=input, output=output, metadata=metadata, + available_skills=available_skills, version=version, level=level, status_message=status_message, @@ -1251,6 +1290,7 @@ def _create_span_with_parent_context( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -1271,6 +1311,7 @@ def _create_span_with_parent_context( input=input, output=output, metadata=metadata, + available_skills=available_skills, version=version, level=level, status_message=status_message, @@ -1299,6 +1340,7 @@ def _start_as_current_otel_span_with_processed_media( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -1335,6 +1377,7 @@ def _start_as_current_otel_span_with_processed_media( "input": input, "output": output, "metadata": metadata, + "available_skills": available_skills, "version": version, "level": level, "status_message": status_message, @@ -1381,6 +1424,7 @@ def update_current_generation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -1402,6 +1446,7 @@ def update_current_generation( input: Updated input data for the model output: Output from the model (e.g., completions) metadata: Additional metadata to associate with the generation + available_skills: Skills available to this observation. version: Version identifier for the model or component level: Importance level of the generation (info, warning, error) status_message: Optional status message for the generation @@ -1448,6 +1493,7 @@ def update_current_generation( input=input, output=output, metadata=metadata, + available_skills=available_skills, version=version, level=level, status_message=status_message, @@ -1466,6 +1512,7 @@ def update_current_span( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -1481,6 +1528,7 @@ def update_current_span( input: Updated input data for the operation output: Output data from the operation metadata: Additional metadata to associate with the span + available_skills: Skills available to this observation. version: Version identifier for the code or component level: Importance level of the span (info, warning, error) status_message: Optional status message for the span @@ -1527,6 +1575,7 @@ def update_current_span( input=input, output=output, metadata=metadata, + available_skills=available_skills, version=version, level=level, status_message=status_message, @@ -1571,6 +1620,7 @@ def create_event( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -1585,6 +1635,7 @@ def create_event( input: Input data for the operation (can be any JSON-serializable object) output: Output data from the operation (can be any JSON-serializable object) metadata: Additional metadata to associate with the span + available_skills: Skills available to this observation. version: Version identifier for the code or component level: Importance level of the span (info, warning, error) status_message: Optional status message for the span @@ -1612,7 +1663,8 @@ def create_event( cast(otel_trace_api.Span, remote_parent_span) ): otel_span = self._otel_tracer.start_span( - name=name, start_time=timestamp + name=name, + start_time=timestamp, ) _set_span_attributes_within_limit( otel_span, {LangfuseOtelSpanAttributes.AS_ROOT: True} @@ -1628,13 +1680,17 @@ def create_event( input=input, output=output, metadata=metadata, + available_skills=available_skills, version=version, level=level, status_message=status_message, ).end(end_time=timestamp), ) - otel_span = self._otel_tracer.start_span(name=name, start_time=timestamp) + otel_span = self._otel_tracer.start_span( + name=name, + start_time=timestamp, + ) return cast( LangfuseEvent, @@ -1646,6 +1702,7 @@ def create_event( input=input, output=output, metadata=metadata, + available_skills=available_skills, version=version, level=level, status_message=status_message, diff --git a/langfuse/_client/get_client.py b/langfuse/_client/get_client.py index eceaed5ef..c1355872f 100644 --- a/langfuse/_client/get_client.py +++ b/langfuse/_client/get_client.py @@ -57,6 +57,7 @@ def _create_client_from_instance( id_generator=instance.id_generator, span_exporter=instance.span_exporter, otel_compression=instance.otel_compression, + skills=instance.skills, httpx_client=instance.httpx_client, ) diff --git a/langfuse/_client/propagation.py b/langfuse/_client/propagation.py index d36ca534a..f71aba269 100644 --- a/langfuse/_client/propagation.py +++ b/langfuse/_client/propagation.py @@ -43,9 +43,11 @@ _serialize_metadata_value, ) from langfuse._client.constants import LANGFUSE_SDK_EXPERIMENT_ENVIRONMENT +from langfuse._client.skills import serialize_available_skills from langfuse._client.span import _set_span_attributes_within_limit from langfuse.logger import langfuse_logger from langfuse.model import PromptClient +from langfuse.types import SkillReference PropagatedKeys = Literal[ "user_id", @@ -57,6 +59,7 @@ "environment", "prompt_name", "prompt_version", + "skills_available", ] InternalPropagatedKeys = Literal[ @@ -79,6 +82,7 @@ "environment", "prompt_name", "prompt_version", + "skills_available", "experiment_id", "experiment_name", "experiment_metadata", @@ -125,6 +129,8 @@ def propagate_attributes( trace_name: Optional[str] = None, environment: Optional[str] = None, prompt: Optional[Union[PromptClient, Mapping[str, Any]]] = None, + skills_available: Optional[List[SkillReference]] = None, + attach_skills: Optional[bool] = None, as_baggage: bool = False, ) -> _AgnosticContextManager[Any]: """Propagate trace-level attributes to all spans created within this context. @@ -185,6 +191,11 @@ def propagate_attributes( takes precedence over the local client default configured via `Langfuse(environment=...)` or `LANGFUSE_TRACING_ENVIRONMENT` for spans created while this propagation context is active. + skills_available: Skill references with name and optional version; replaces + skills on the current span and in inherited context, including when empty. + Never sent as baggage. + attach_skills: Attach the configured skill catalog to spans + created in this context; disabled by default. Never sent as baggage. as_baggage: If True, propagates attributes using OpenTelemetry baggage for cross-process/service propagation. **Security warning**: When enabled, attribute values are added to HTTP headers on ALL outbound requests. @@ -314,6 +325,8 @@ def propagate_attributes( trace_name=trace_name, environment=environment, prompt=prompt, + skills_available=skills_available, + attach_skills=attach_skills, as_baggage=as_baggage, ) @@ -329,12 +342,30 @@ def _propagate_attributes( trace_name: Optional[str] = None, environment: Optional[str] = None, prompt: Optional[Union[PromptClient, Mapping[str, Any]]] = None, + skills_available: Optional[List[SkillReference]] = None, + attach_skills: Optional[bool] = None, as_baggage: bool = False, experiment: Optional[PropagatedExperimentAttributes] = None, ) -> Generator[Any, Any, Any]: context = otel_context_api.get_current() current_span = otel_trace_api.get_current_span() + serialized_skills = serialize_available_skills(skills_available) + if serialized_skills is not None: + context = _set_propagated_attribute( + key="skills_available", + value=serialized_skills, + context=context, + span=current_span, + as_baggage=False, + ) + if isinstance(attach_skills, bool): + context = otel_context_api.set_value( + key=_get_propagated_context_key("attach_skills"), + value=attach_skills, + context=context, + ) + propagated_string_attributes: Dict[str, Optional[Union[str, List[str]]]] = { "user_id": user_id, "session_id": session_id, @@ -785,6 +816,7 @@ def _get_propagated_span_key(key: str) -> str: "metadata": LangfuseOtelSpanAttributes.TRACE_METADATA, "prompt_name": LangfuseOtelSpanAttributes.OBSERVATION_PROMPT_NAME, "prompt_version": LangfuseOtelSpanAttributes.OBSERVATION_PROMPT_VERSION, + "skills_available": LangfuseOtelSpanAttributes.OBSERVATION_SKILLS_AVAILABLE, "experiment_id": LangfuseOtelSpanAttributes.EXPERIMENT_ID, "experiment_name": LangfuseOtelSpanAttributes.EXPERIMENT_NAME, "experiment_metadata": LangfuseOtelSpanAttributes.EXPERIMENT_METADATA, diff --git a/langfuse/_client/resource_manager.py b/langfuse/_client/resource_manager.py index a274075e7..1c1afa424 100644 --- a/langfuse/_client/resource_manager.py +++ b/langfuse/_client/resource_manager.py @@ -48,7 +48,7 @@ from langfuse._utils.prompt_cache import PromptCache from langfuse.api import AsyncLangfuseAPI, LangfuseAPI from langfuse.logger import langfuse_logger -from langfuse.types import MaskFunction, MaskOtelSpansFunction +from langfuse.types import MaskFunction, MaskOtelSpansFunction, SkillsConfig from .._version import __version__ as langfuse_version @@ -132,6 +132,7 @@ def __new__( id_generator: Optional[IdGenerator] = None, span_exporter: Optional[SpanExporter] = None, otel_compression: Optional[Literal["gzip", "none"]] = None, + skills: Optional[SkillsConfig] = None, ) -> "LangfuseResourceManager": if public_key in cls._instances: return cls._instances[public_key] @@ -170,6 +171,7 @@ def __new__( id_generator=id_generator, span_exporter=span_exporter, otel_compression=otel_compression, + skills=skills, ) cls._instances[public_key] = instance @@ -199,6 +201,7 @@ def _initialize_instance( id_generator: Optional[IdGenerator] = None, span_exporter: Optional[SpanExporter] = None, otel_compression: Optional[Literal["gzip", "none"]] = None, + skills: Optional[SkillsConfig] = None, ) -> None: self.public_key = public_key self.secret_key = secret_key @@ -221,6 +224,7 @@ def _initialize_instance( self.id_generator = id_generator self.span_exporter = span_exporter self.otel_compression = otel_compression + self.skills = skills self.tracer_provider: Optional[TracerProvider] = None self._custom_httpx_client = httpx_client @@ -260,6 +264,7 @@ def _initialize_instance( media_manager=self._media_manager, mask_otel_spans=mask_otel_spans, otel_compression=otel_compression, + skills=skills, ) tracer_provider.add_span_processor(langfuse_processor) diff --git a/langfuse/_client/skills.py b/langfuse/_client/skills.py new file mode 100644 index 000000000..64f5d5884 --- /dev/null +++ b/langfuse/_client/skills.py @@ -0,0 +1,137 @@ +"""Local skill discovery and availability configuration.""" + +import json +import os +import stat +from pathlib import Path +from typing import Any, Optional + +from langfuse.logger import langfuse_logger +from langfuse.types import SkillReference + +_CLI_LOCK = "langfuse-skills.lock" + + +def discover_skills(*, dir: str) -> list[SkillReference]: + """Read immediate skill directories and optional CLI installation metadata. + + Directories must contain a regular SKILL.md file. Hidden directories are + skipped and symlinks are followed. The configured directory's + langfuse-skills.lock takes precedence over the working directory's lock. + Unlocked skills use their directory name without a version. Lock records + do not verify local edits. + + Missing directories return an empty list. Unreadable or malformed lockfiles + warn and fall back to directory names. Entries that cannot be inspected, + invalid lock entries, and duplicate names are skipped with a warning. + """ + root = Path(os.path.abspath(dir)) + if _stat_if_exists(root) is None: + return [] + + locked: dict[Path, SkillReference] = {} + try: + locked = _read_cli_lock(_find_cli_lock(root)) + except (OSError, ValueError) as error: + langfuse_logger.warning( + "Could not read skill lockfile for %s. Using directory names without versions: %s", + root, + error, + ) + available: dict[str, SkillReference] = {} + with os.scandir(root) as entries: + candidates = sorted(entries, key=lambda entry: entry.name) + for candidate in candidates: + if candidate.name.startswith("."): + continue + directory = Path(candidate.path) + try: + directory_stat = _stat_if_exists(directory) + if directory_stat is None or not stat.S_ISDIR(directory_stat.st_mode): + continue + skill_file = _stat_if_exists(directory / "SKILL.md") + if skill_file is None or not stat.S_ISREG(skill_file.st_mode): + continue + except OSError as error: + langfuse_logger.warning( + "Skipping skill directory %s: could not inspect files: %s", + directory, + error, + ) + continue + skill = locked.get(directory, {"name": candidate.name}) + name = skill["name"] + if len(name) > 128: + langfuse_logger.warning( + "Skipping skill %s: name exceeds 128 characters.", name + ) + continue + if name in available: + langfuse_logger.warning("Skipping duplicate skill name %s.", name) + continue + available[name] = skill + return list(available.values()) + + +def _stat_if_exists(path: Path) -> Optional[os.stat_result]: + try: + return path.stat() + except FileNotFoundError: + return None + + +def _find_cli_lock(directory: Path) -> Optional[Path]: + for path in (directory / _CLI_LOCK, Path.cwd() / _CLI_LOCK): + if _stat_if_exists(path) is not None: + return path + return None + + +def _read_cli_lock(file: Optional[Path]) -> dict[Path, SkillReference]: + if file is None: + return {} + file_stat = _stat_if_exists(file) + if file_stat is None: + return {} + if not stat.S_ISREG(file_stat.st_mode): + message = f"Invalid skill lockfile {file}." + raise ValueError(message) + value = json.loads(file.read_text(encoding="utf-8")) + if ( + not isinstance(value, dict) + or type(value.get("lockVersion")) is not int + or not isinstance(value.get("skills"), list) + ): + message = f"Unsupported or invalid skill lockfile {file}." + raise ValueError(message) + locked: dict[Path, SkillReference] = {} + for skill in value["skills"]: + if not _is_valid_locked_skill(skill): + langfuse_logger.warning("Skipping invalid skill entry in %s.", file) + continue + path = Path(os.path.abspath(file.parent / skill["path"])) + locked[path] = {"name": skill["name"], "version": skill["version"]} + return locked + + +def _is_valid_locked_skill(skill: Any) -> bool: + return ( + isinstance(skill, dict) + and isinstance(skill.get("path"), str) + and bool(skill["path"]) + and isinstance(skill.get("name"), str) + and 0 < len(skill["name"]) <= 128 + and type(skill.get("version")) is int + and 1 <= skill["version"] <= 2_147_483_647 + ) + + +def serialize_available_skills(value: Any) -> Optional[str]: + """Serialize skill attributes without projecting away unknown fields.""" + if value is None: + return None + try: + return json.dumps(value, separators=(",", ":"), allow_nan=False) + except Exception: + langfuse_logger.warning("Failed to serialize available skills. Dropping value.") + return None diff --git a/langfuse/_client/span.py b/langfuse/_client/span.py index 725f1aac0..622df4db4 100644 --- a/langfuse/_client/span.py +++ b/langfuse/_client/span.py @@ -56,7 +56,7 @@ from langfuse._client.utils import get_string_span_attribute from langfuse.api import MapValue, ScoreDataType from langfuse.logger import langfuse_logger -from langfuse.types import SpanLevel +from langfuse.types import SkillReference, SpanLevel # Factory mapping for observation classes # Note: "event" is handled separately due to special instantiation logic @@ -90,6 +90,7 @@ def _is_metadata_key(key: str) -> bool: LangfuseOtelSpanAttributes.ENVIRONMENT, LangfuseOtelSpanAttributes.OBSERVATION_INPUT, LangfuseOtelSpanAttributes.OBSERVATION_OUTPUT, + LangfuseOtelSpanAttributes.OBSERVATION_SKILLS_AVAILABLE, ) _RESERVED_OBSERVATION_ATTRIBUTE_KEYS: Tuple[str, ...] = ( *_RESERVED_SPAN_ATTRIBUTE_KEYS, @@ -276,6 +277,7 @@ def __init__( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, environment: Optional[str] = None, release: Optional[str] = None, version: Optional[str] = None, @@ -297,6 +299,7 @@ def __init__( input: Input data for the span (any JSON-serializable object) output: Output data from the span (any JSON-serializable object) metadata: Additional metadata to associate with the span + available_skills: Skills available to this observation. environment: Local tracing environment fallback. If the underlying OpenTelemetry span already has `langfuse.environment` from propagated context or baggage, that propagated value takes @@ -365,6 +368,7 @@ def __init__( input=media_processed_input, output=media_processed_output, metadata=media_processed_metadata, + available_skills=available_skills, version=version, level=level, status_message=status_message, @@ -386,6 +390,7 @@ def __init__( input=media_processed_input, output=media_processed_output, metadata=media_processed_metadata, + available_skills=available_skills, version=version, level=level, status_message=status_message, @@ -819,6 +824,7 @@ def update( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -840,6 +846,7 @@ def update( input: Updated input data for the operation output: Output data from the operation metadata: Additional metadata to associate with the observation + available_skills: Skills available to this observation. version: Version identifier for the code or component level: Importance level of the observation (info, warning, error) status_message: Optional status message for the observation @@ -874,6 +881,7 @@ def update( input=processed_input, output=processed_output, metadata=processed_metadata, + available_skills=available_skills, version=version, level=level, status_message=status_message, @@ -894,6 +902,7 @@ def update( input=processed_input, output=processed_output, metadata=processed_metadata, + available_skills=available_skills, version=version, level=level, status_message=status_message, @@ -926,6 +935,7 @@ def start_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -940,6 +950,7 @@ def start_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -960,6 +971,7 @@ def start_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -974,6 +986,7 @@ def start_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -988,6 +1001,7 @@ def start_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -1002,6 +1016,7 @@ def start_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -1016,6 +1031,7 @@ def start_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -1030,6 +1046,7 @@ def start_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -1050,6 +1067,7 @@ def start_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -1064,6 +1082,7 @@ def start_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -1077,6 +1096,7 @@ def start_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -1110,6 +1130,7 @@ def start_observation( input: Input data for the operation output: Output data from the operation metadata: Additional metadata to associate with the observation + available_skills: Skills available to this observation. version: Version identifier for the code or component level: Importance level of the observation (info, warning, error) status_message: Optional status message for the observation @@ -1137,6 +1158,7 @@ def start_observation( input=input, output=output, metadata=metadata, + available_skills=available_skills, environment=self._environment, release=self._release, version=version, @@ -1153,7 +1175,9 @@ def start_observation( observation_class = LangfuseSpan with otel_trace_api.use_span(self._otel_span): - new_otel_span = self._langfuse_client._otel_tracer.start_span(name=name) + new_otel_span = self._langfuse_client._otel_tracer.start_span( + name=name, + ) common_args = { "otel_span": new_otel_span, @@ -1163,6 +1187,7 @@ def start_observation( "input": input, "output": output, "metadata": metadata, + "available_skills": available_skills, "version": version, "level": level, "status_message": status_message, @@ -1191,6 +1216,7 @@ def start_as_current_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -1205,6 +1231,7 @@ def start_as_current_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -1225,6 +1252,7 @@ def start_as_current_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -1245,6 +1273,7 @@ def start_as_current_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -1259,6 +1288,7 @@ def start_as_current_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -1273,6 +1303,7 @@ def start_as_current_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -1287,6 +1318,7 @@ def start_as_current_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -1301,6 +1333,7 @@ def start_as_current_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -1315,6 +1348,7 @@ def start_as_current_observation( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -1328,6 +1362,7 @@ def start_as_current_observation( # type: ignore[misc] input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -1362,6 +1397,7 @@ def start_as_current_observation( # type: ignore[misc] input: Input data for the operation output: Output data from the operation metadata: Additional metadata to associate with the observation + available_skills: Skills available to this observation. version: Version identifier for the code or component level: Importance level of the observation (info, warning, error) status_message: Optional status message for the observation @@ -1383,6 +1419,7 @@ def start_as_current_observation( # type: ignore[misc] input=input, output=output, metadata=metadata, + available_skills=available_skills, version=version, level=level, status_message=status_message, @@ -1401,6 +1438,7 @@ def create_event( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, @@ -1412,6 +1450,7 @@ def create_event( input: Input data for the operation (can be any JSON-serializable object) output: Output data from the operation (can be any JSON-serializable object) metadata: Additional metadata to associate with the span + available_skills: Skills available to this observation. version: Version identifier for the code or component level: Importance level of the span (info, warning, error) status_message: Optional status message for the span @@ -1428,7 +1467,8 @@ def create_event( with otel_trace_api.use_span(self._otel_span): new_otel_span = self._langfuse_client._otel_tracer.start_span( - name=name, start_time=timestamp + name=name, + start_time=timestamp, ) return cast( @@ -1439,6 +1479,7 @@ def create_event( input=input, output=output, metadata=metadata, + available_skills=available_skills, environment=self._environment, release=self._release, version=version, @@ -1466,6 +1507,7 @@ def __init__( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, environment: Optional[str] = None, release: Optional[str] = None, version: Optional[str] = None, @@ -1480,6 +1522,7 @@ def __init__( input: Input data for the span (any JSON-serializable object) output: Output data from the span (any JSON-serializable object) metadata: Additional metadata to associate with the span + available_skills: Skills available to this observation. environment: The tracing environment release: Release identifier for the application version: Version identifier for the code or component @@ -1493,6 +1536,7 @@ def __init__( input=input, output=output, metadata=metadata, + available_skills=available_skills, environment=environment, release=release, version=version, @@ -1517,6 +1561,7 @@ def __init__( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, environment: Optional[str] = None, release: Optional[str] = None, version: Optional[str] = None, @@ -1537,6 +1582,7 @@ def __init__( input: Input data for the generation (e.g., prompts) output: Output from the generation (e.g., completions) metadata: Additional metadata to associate with the generation + available_skills: Skills available to this observation. environment: The tracing environment release: Release identifier for the application version: Version identifier for the model or component @@ -1556,6 +1602,7 @@ def __init__( input=input, output=output, metadata=metadata, + available_skills=available_skills, environment=environment, release=release, version=version, @@ -1581,6 +1628,7 @@ def __init__( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, environment: Optional[str] = None, release: Optional[str] = None, version: Optional[str] = None, @@ -1595,6 +1643,7 @@ def __init__( input: Input data for the event output: Output from the event metadata: Additional metadata to associate with the generation + available_skills: Skills available to this observation. environment: The tracing environment release: Release identifier for the application version: Version identifier for the model or component @@ -1608,6 +1657,7 @@ def __init__( input=input, output=output, metadata=metadata, + available_skills=available_skills, environment=environment, release=release, version=version, @@ -1622,6 +1672,7 @@ def update( input: Optional[Any] = None, output: Optional[Any] = None, metadata: Optional[Any] = None, + available_skills: Optional[list[SkillReference]] = None, version: Optional[str] = None, level: Optional[SpanLevel] = None, status_message: Optional[str] = None, diff --git a/langfuse/_client/span_processor.py b/langfuse/_client/span_processor.py index 438551fd6..b6ebbbce6 100644 --- a/langfuse/_client/span_processor.py +++ b/langfuse/_client/span_processor.py @@ -40,7 +40,9 @@ from langfuse._client.propagation import ( _get_langfuse_trace_id_from_baggage, _get_propagated_attributes_from_context, + _get_propagated_context_key, ) +from langfuse._client.skills import serialize_available_skills from langfuse._client.span import _set_span_attributes_within_limit from langfuse._client.span_exporter import LangfuseTransformingSpanExporter from langfuse._client.span_filter import ( @@ -52,7 +54,7 @@ from langfuse._task_manager.media_manager import MediaManager from langfuse._version import __version__ as langfuse_version from langfuse.logger import langfuse_logger -from langfuse.types import MaskOtelSpansFunction +from langfuse.types import MaskOtelSpansFunction, SkillsConfig def _resolve_max_batch_size_bytes() -> Optional[int]: @@ -157,7 +159,13 @@ def __init__( media_manager: Optional[MediaManager] = None, mask_otel_spans: Optional[MaskOtelSpansFunction] = None, otel_compression: Optional[Literal["gzip", "none"]] = None, + skills: Optional[SkillsConfig] = None, ): + self._available_skills = ( + serialize_available_skills(skills["available"]) + if skills is not None + else None + ) self.public_key = public_key self._should_export_span = should_export_span or is_default_export_span @@ -243,6 +251,21 @@ def on_start(self, span: Span, parent_context: Optional[Context] = None) -> None ) } + skills_key = LangfuseOtelSpanAttributes.OBSERVATION_SKILLS_AVAILABLE + # Existing span skills take precedence over propagation, then configured skills. + if skills_key in existing_attributes: + propagated_attributes.pop(skills_key, None) + elif ( + skills_key not in propagated_attributes + and self._available_skills is not None + and (not is_langfuse_span(span) or self._is_langfuse_project_span(span)) + and context_api.get_value( + _get_propagated_context_key("attach_skills"), context=context + ) + is True + ): + propagated_attributes[skills_key] = self._available_skills + if propagated_attributes: _set_span_attributes_within_limit(span, propagated_attributes) diff --git a/langfuse/types.py b/langfuse/types.py index 991b1e0a3..3a6935486 100644 --- a/langfuse/types.py +++ b/langfuse/types.py @@ -340,6 +340,19 @@ class ParsedMediaReference(TypedDict): content_type: MediaContentType +class SkillReference(TypedDict): + """A skill available to an agent, optionally pinned to a version.""" + + name: str + version: NotRequired[int] + + +class SkillsConfig(TypedDict): + """Skills to attach when propagation enables attach_skills.""" + + available: list[SkillReference] + + class TraceContext(TypedDict): trace_id: str parent_span_id: NotRequired[str] @@ -358,4 +371,6 @@ class TraceContext(TypedDict): "OtelSpanPatch", "ParsedMediaReference", "TraceContext", + "SkillReference", + "SkillsConfig", ] diff --git a/tests/conftest.py b/tests/conftest.py index aa4896b34..ae437faf4 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -131,6 +131,7 @@ def mock_init(self: Any, **kwargs: Any) -> None: ) self._app_root_lock = threading.Lock() self._span_export_expectation_by_id = {} + self._available_skills = None BatchSpanProcessor.__init__( self, span_exporter=memory_exporter, diff --git a/tests/unit/test_metadata_attribute_limit.py b/tests/unit/test_metadata_attribute_limit.py index e508bd147..398707a32 100644 --- a/tests/unit/test_metadata_attribute_limit.py +++ b/tests/unit/test_metadata_attribute_limit.py @@ -340,8 +340,8 @@ def test_reserved_keys_already_written_cost_no_extra_slot(small_limit_client, ge def test_reserved_keys_depend_on_observation_type(langfuse_memory_client, get_span): - assert len(_RESERVED_SPAN_ATTRIBUTE_KEYS) == 7 - assert len(_RESERVED_OBSERVATION_ATTRIBUTE_KEYS) == 14 + assert len(_RESERVED_SPAN_ATTRIBUTE_KEYS) == 8 + assert len(_RESERVED_OBSERVATION_ATTRIBUTE_KEYS) == 15 for observation_type in ("generation", "embedding", None): assert ( _reserved_attribute_keys(observation_type) @@ -375,8 +375,8 @@ def test_reserved_keys_depend_on_observation_type(langfuse_memory_client, get_sp generation = get_span("typed-generation") span_count = len(_metadata_keys(span)) generation_count = len(_metadata_keys(generation)) - # The span reserves 5 open slots (level, status message, version, output, - # plus environment when unset) instead of 12 for the generation. + # The span reserves 6 open slots (level, status message, version, output, + # skills, plus environment when unset) instead of 13 for the generation. assert span_count - generation_count == 7 assert span.dropped_attributes == 0 assert generation.dropped_attributes == 0 @@ -674,3 +674,21 @@ def task(**kwargs): warnings = _limit_warnings(caplog) assert warnings assert all("other attribute(s)" not in w.getMessage() for w in warnings) + + +def test_metadata_reserves_room_for_later_skill_availability( + small_limit_client, get_span +): + observation = small_limit_client.start_observation( + name="late-skills", metadata={f"key_{i}": i for i in range(130)} + ) + before = dict(observation._otel_span.attributes) + observation.update(available_skills=[{"name": "support", "version": 2}]) + observation.end() + small_limit_client.flush() + + attributes = get_span("late-skills").attributes + assert all(attributes[key] == value for key, value in before.items()) + assert json.loads( + attributes[LangfuseOtelSpanAttributes.OBSERVATION_SKILLS_AVAILABLE] + ) == [{"name": "support", "version": 2}] diff --git a/tests/unit/test_otel.py b/tests/unit/test_otel.py index b49a0b04d..f5757fa15 100644 --- a/tests/unit/test_otel.py +++ b/tests/unit/test_otel.py @@ -197,6 +197,7 @@ def mock_init(self, **kwargs): ) self._app_root_lock = threading.Lock() self._span_export_expectation_by_id = {} + self._available_skills = None BatchSpanProcessor.__init__( self, span_exporter=memory_exporter, @@ -2456,6 +2457,7 @@ def mock_processor_init(self, **kwargs): ) self._app_root_lock = threading.Lock() self._span_export_expectation_by_id = {} + self._available_skills = None # Use the appropriate exporter based on the project key if self.public_key == project1_key: exporter = exporter_project1 @@ -2835,6 +2837,7 @@ def mock_processor_init(self, **kwargs): ) self._app_root_lock = threading.Lock() self._span_export_expectation_by_id = {} + self._available_skills = None # For testing, use the appropriate exporter based on setup exporter = kwargs.get("_test_exporter", blocked_exporter) @@ -3931,3 +3934,109 @@ def test_langfuse_event_update_immutability(self, langfuse_client, caplog): assert result is event parent_span.end() + + +@pytest.mark.parametrize("as_type", ["span", "generation", "tool", "embedding"]) +@pytest.mark.parametrize("parented", [False, True]) +@pytest.mark.parametrize("current", [False, True]) +def test_observation_available_skills_creation( + langfuse_memory_client, get_span, as_type, parented, current +): + skills = [{"name": "support", "version": 2, "future_field": {"enabled": True}}] + parent = langfuse_memory_client.start_observation(name="parent") + owner = parent if parented else langfuse_memory_client + if current: + with owner.start_as_current_observation( + name="skills", as_type=as_type, available_skills=skills + ): + pass + else: + owner.start_observation( + name="skills", as_type=as_type, available_skills=skills + ).end() + parent.end() + langfuse_memory_client.flush() + + assert ( + json.loads( + get_span("skills").attributes[ + LangfuseOtelSpanAttributes.OBSERVATION_SKILLS_AVAILABLE + ] + ) + == skills + ) + + +@pytest.mark.parametrize("current", [False, True]) +def test_remote_observation_available_skills(langfuse_memory_client, get_span, current): + kwargs = { + "name": "remote-skills", + "as_type": "generation", + "trace_context": {"trace_id": "12345678901234567890123456789012"}, + "available_skills": [], + } + if current: + with langfuse_memory_client.start_as_current_observation(**kwargs): + pass + else: + langfuse_memory_client.start_observation(**kwargs).end() + langfuse_memory_client.flush() + + assert ( + get_span("remote-skills").attributes[ + LangfuseOtelSpanAttributes.OBSERVATION_SKILLS_AVAILABLE + ] + == "[]" + ) + + +@pytest.mark.parametrize("parented", [False, True]) +def test_event_available_skills(langfuse_memory_client, get_span, parented): + parent = langfuse_memory_client.start_observation(name="parent") + owner = parent if parented else langfuse_memory_client + owner.create_event(name="skill-event", available_skills=[{"name": "support"}]) + if parented: + parent.start_observation( + name="other-event", as_type="event", available_skills=[] + ) + parent.end() + langfuse_memory_client.flush() + + assert json.loads( + get_span("skill-event").attributes[ + LangfuseOtelSpanAttributes.OBSERVATION_SKILLS_AVAILABLE + ] + ) == [{"name": "support"}] + if parented: + assert ( + get_span("other-event").attributes[ + LangfuseOtelSpanAttributes.OBSERVATION_SKILLS_AVAILABLE + ] + == "[]" + ) + + +@pytest.mark.parametrize("as_type", ["span", "generation"]) +def test_observation_available_skills_updates( + langfuse_memory_client, get_span, as_type +): + key = LangfuseOtelSpanAttributes.OBSERVATION_SKILLS_AVAILABLE + with langfuse_memory_client.start_as_current_observation( + name="updates", as_type=as_type, available_skills=[{"name": "initial"}] + ) as observation: + observation.update(available_skills=[{"name": "replacement", "version": 3}]) + assert json.loads(observation._otel_span.attributes[key]) == [ + {"name": "replacement", "version": 3} + ] + observation.update(output="unrelated") + assert json.loads(observation._otel_span.attributes[key]) == [ + {"name": "replacement", "version": 3} + ] + update_current = ( + langfuse_memory_client.update_current_generation + if as_type == "generation" + else langfuse_memory_client.update_current_span + ) + update_current(available_skills=[]) + langfuse_memory_client.flush() + assert get_span("updates").attributes[key] == "[]" diff --git a/tests/unit/test_propagate_attributes.py b/tests/unit/test_propagate_attributes.py index c1b5947fe..5381d76da 100644 --- a/tests/unit/test_propagate_attributes.py +++ b/tests/unit/test_propagate_attributes.py @@ -10,12 +10,14 @@ from datetime import datetime import pytest +from opentelemetry import baggage, context from opentelemetry.instrumentation.threading import ThreadingInstrumentor from langfuse import propagate_attributes from langfuse._client.attributes import LangfuseOtelSpanAttributes, _serialize from langfuse._client.constants import LANGFUSE_SDK_EXPERIMENT_ENVIRONMENT from langfuse._client.datasets import DatasetClient +from langfuse._client.propagation import _get_propagated_attributes_from_context from langfuse.api import Dataset, DatasetItem, DatasetStatus from tests.unit.test_otel import TestOTelBase @@ -3831,3 +3833,90 @@ def test_prompt_composes_with_outer_propagated_attributes( self.verify_missing_attribute( after_span, LangfuseOtelSpanAttributes.OBSERVATION_PROMPT_NAME ) + + +class TestPropagatedSkills(TestPropagateAttributesBase): + def test_catalog_replaces_inherited_and_updates_current_span( + self, langfuse_client, memory_exporter + ): + outer = [{"name": "refund", "version": 2, "future_field": {"enabled": True}}] + with propagate_attributes(skills_available=outer): + with langfuse_client.start_as_current_observation(name="parent"): + with propagate_attributes(skills_available=[]): + langfuse_client.start_observation(name="cleared").end() + langfuse_client.start_observation(name="restored").end() + langfuse_client.start_observation(name="outside").end() + for name in ("parent", "cleared"): + _assert_json_attribute( + self.get_span_by_name(memory_exporter, name), + LangfuseOtelSpanAttributes.OBSERVATION_SKILLS_AVAILABLE, + [], + ) + _assert_json_attribute( + self.get_span_by_name(memory_exporter, "restored"), + LangfuseOtelSpanAttributes.OBSERVATION_SKILLS_AVAILABLE, + outer, + ) + self.verify_missing_attribute( + self.get_span_by_name(memory_exporter, "outside"), + LangfuseOtelSpanAttributes.OBSERVATION_SKILLS_AVAILABLE, + ) + + @pytest.mark.parametrize("skills", [[{"name": "refund"}], []]) + @pytest.mark.parametrize("set_on_update", [False, True]) + def test_propagation_updates_current_observation_skills( + self, langfuse_client, memory_exporter, skills, set_on_update + ): + inherited = [{"name": "child"}] + with propagate_attributes(skills_available=[{"name": "outer"}]): + with langfuse_client.start_as_current_observation( + name="parent", available_skills=None if set_on_update else skills + ) as parent: + if set_on_update: + parent.update(available_skills=skills) + assert ( + json.loads( + parent._otel_span.attributes[ + LangfuseOtelSpanAttributes.OBSERVATION_SKILLS_AVAILABLE + ] + ) + == skills + ) + with propagate_attributes(skills_available=inherited): + langfuse_client.start_observation(name="child").end() + for name, expected in (("parent", inherited), ("child", inherited)): + _assert_json_attribute( + self.get_span_by_name(memory_exporter, name), + LangfuseOtelSpanAttributes.OBSERVATION_SKILLS_AVAILABLE, + expected, + ) + + def test_skills_baggage_is_read_but_not_written(self): + with propagate_attributes( + skills_available=[{"name": "local"}], + attach_skills=True, + as_baggage=True, + ): + assert not any("skills" in key for key in baggage.get_all()) + incoming = baggage.set_baggage( + "langfuse_skills_available", '[{"name":"remote"}]' + ) + assert _get_propagated_attributes_from_context(incoming) == { + LangfuseOtelSpanAttributes.OBSERVATION_SKILLS_AVAILABLE: '[{"name":"remote"}]' + } + assert not _get_propagated_attributes_from_context(context.get_current()) + + def test_invalid_serialization_keeps_outer_catalog( + self, langfuse_client, memory_exporter, caplog + ): + with propagate_attributes(skills_available=[{"name": "outer"}]): + with propagate_attributes( + skills_available=[{"name": "bad", "extra": object()}] + ): + langfuse_client.start_observation(name="child").end() + _assert_json_attribute( + self.get_span_by_name(memory_exporter, "child"), + LangfuseOtelSpanAttributes.OBSERVATION_SKILLS_AVAILABLE, + [{"name": "outer"}], + ) + assert "Failed to serialize available skills" in caplog.text diff --git a/tests/unit/test_resource_manager.py b/tests/unit/test_resource_manager.py index 27e600cdc..788dc2ced 100644 --- a/tests/unit/test_resource_manager.py +++ b/tests/unit/test_resource_manager.py @@ -59,6 +59,7 @@ def mask_otel_spans(*, params): "mask_otel_spans": mask_otel_spans, "additional_headers": {"X-Custom": "value"}, "span_exporter": span_exporter, + "skills": {"available": [{"name": "support", "version": 2}]}, } original_client = Langfuse(**settings) @@ -76,6 +77,7 @@ def mask_otel_spans(*, params): assert rm.mask_otel_spans is mask_otel_spans assert rm.additional_headers == settings["additional_headers"] assert rm.span_exporter is span_exporter + assert rm.skills is settings["skills"] original_client.shutdown() diff --git a/tests/unit/test_skills.py b/tests/unit/test_skills.py new file mode 100644 index 000000000..9b3d98cc3 --- /dev/null +++ b/tests/unit/test_skills.py @@ -0,0 +1,163 @@ +import json + +import pytest + +from langfuse._client.skills import ( + discover_skills, + serialize_available_skills, +) + + +def _install(root, name): + directory = root / name + directory.mkdir(parents=True) + (directory / "SKILL.md").write_text("# Skill", encoding="utf-8") + return directory + + +def _write_lock(root, skills, lock_version=1): + (root / "langfuse-skills.lock").write_text( + json.dumps({"lockVersion": lock_version, "skills": skills}), encoding="utf-8" + ) + + +def test_discovery_follows_symlinks_and_only_includes_immediate_skills( + tmp_path, monkeypatch +): + monkeypatch.chdir(tmp_path) + root = tmp_path / "skills" + _install(root, "plain") + _install(root, ".hidden") + _install(root, "nested/deeper") + (root / "directory-link").symlink_to(root / "plain", target_is_directory=True) + (root / "file-link").mkdir() + (root / "file-link" / "SKILL.md").symlink_to(root / "plain" / "SKILL.md") + (root / "directory-marker" / "SKILL.md").mkdir(parents=True) + + assert discover_skills(dir=str(root)) == [ + {"name": "directory-link"}, + {"name": "file-link"}, + {"name": "plain"}, + ] + assert discover_skills(dir=str(tmp_path / "missing")) == [] + + +@pytest.mark.parametrize("lock_version", [1, 2]) +def test_configured_lock_precedes_cwd_and_resolves_paths_from_lock( + tmp_path, monkeypatch, lock_version +): + monkeypatch.chdir(tmp_path) + root = tmp_path / "skills" + _install(root, "customer") + _write_lock(tmp_path, [{"path": "skills/customer", "name": "cwd", "version": 1}]) + _write_lock( + root, + [{"path": "customer", "name": "configured", "version": 2}], + lock_version=lock_version, + ) + + assert discover_skills(dir="skills") == [{"name": "configured", "version": 2}] + (root / "langfuse-skills.lock").unlink() + assert discover_skills(dir="skills") == [{"name": "cwd", "version": 1}] + + +def test_invalid_entries_duplicates_and_long_names_preserve_other_skills( + tmp_path, monkeypatch, caplog +): + monkeypatch.chdir(tmp_path) + root = tmp_path / "skills" + for name in ("first", "second", "invalid-version", "x" * 129): + _install(root, name) + _write_lock( + root, + [ + {"path": "first", "name": "same", "version": 2_147_483_647}, + {"path": "second", "name": "same", "version": 2}, + {"path": "invalid-version", "name": "bad", "version": True}, + {"path": "first", "name": "bad", "version": 2_147_483_648}, + {"path": "first", "name": "bad", "version": 0}, + {"path": "first", "name": "bad", "version": 1.5}, + ], + ) + + assert discover_skills(dir=str(root)) == [ + {"name": "same", "version": 2_147_483_647}, + {"name": "invalid-version"}, + ] + assert "Skipping invalid skill entry" in caplog.text + assert "Skipping duplicate skill name same" in caplog.text + assert "name exceeds 128 characters" in caplog.text + + +@pytest.mark.parametrize( + "content", + ["not json", '{"lockVersion":1,"skills":{}}', '{"lockVersion":true,"skills":[]}'], +) +def test_malformed_lock_falls_back_to_directory_names( + tmp_path, monkeypatch, content, caplog +): + monkeypatch.chdir(tmp_path) + root = tmp_path / "skills" + _install(root, "customer") + _write_lock(tmp_path, [{"path": "skills/customer", "name": "cwd", "version": 1}]) + (root / "langfuse-skills.lock").write_text(content, encoding="utf-8") + + assert discover_skills(dir=str(root)) == [{"name": "customer"}] + assert "Could not read skill lockfile" in caplog.text + + +def test_discovery_follows_lock_symlink(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + root = tmp_path / "skills" + _install(root, "customer") + _write_lock(tmp_path, [{"path": "customer", "name": "locked", "version": 2}]) + (root / "langfuse-skills.lock").symlink_to(tmp_path / "langfuse-skills.lock") + + assert discover_skills(dir=str(root)) == [{"name": "locked", "version": 2}] + + +@pytest.mark.parametrize("kind", ["directory", "symlink-loop"]) +def test_unreadable_lock_falls_back_to_directory_names( + tmp_path, monkeypatch, kind, caplog +): + monkeypatch.chdir(tmp_path) + _install(tmp_path, "customer") + lock = tmp_path / "langfuse-skills.lock" + if kind == "directory": + lock.mkdir() + else: + lock.symlink_to(lock) + + assert discover_skills(dir=str(tmp_path)) == [{"name": "customer"}] + assert "Could not read skill lockfile" in caplog.text + + +@pytest.mark.parametrize("broken_path", ["directory", "SKILL.md"]) +def test_discovery_skips_entries_with_filesystem_errors( + tmp_path, monkeypatch, broken_path, caplog +): + monkeypatch.chdir(tmp_path) + _install(tmp_path, "valid") + broken = tmp_path / "broken" + if broken_path == "SKILL.md": + broken.mkdir() + broken = broken / "SKILL.md" + broken.symlink_to(broken) + (tmp_path / "missing-target").symlink_to(tmp_path / "missing") + + assert discover_skills(dir=str(tmp_path)) == [{"name": "valid"}] + assert "Skipping skill directory" in caplog.text + assert "missing-target" not in caplog.text + + +def test_serialization_preserves_extra_fields(): + skills = [{"name": "customer", "version": 2, "futureField": {"nested": True}}] + assert json.loads(serialize_available_skills(skills)) == skills + assert serialize_available_skills([]) == "[]" + + +def test_invalid_serialization_does_not_break_tracing(caplog): + cycle = [] + cycle.append(cycle) + assert serialize_available_skills(cycle) is None + assert "Failed to serialize available skills" in caplog.text diff --git a/tests/unit/test_span_processor.py b/tests/unit/test_span_processor.py index d7c5366be..97d53ff58 100644 --- a/tests/unit/test_span_processor.py +++ b/tests/unit/test_span_processor.py @@ -1,4 +1,5 @@ import gzip +import json import logging import threading from http.server import BaseHTTPRequestHandler, HTTPServer @@ -14,7 +15,7 @@ OTEL_EXPORTER_OTLP_COMPRESSION, OTEL_EXPORTER_OTLP_TRACES_COMPRESSION, ) -from opentelemetry.sdk.trace import ReadableSpan, TracerProvider +from opentelemetry.sdk.trace import ReadableSpan, SpanLimits, TracerProvider from opentelemetry.sdk.trace.export import ( SimpleSpanProcessor, SpanExporter, @@ -25,6 +26,7 @@ ) import langfuse._client.span_processor as span_processor_module +from langfuse import LangfuseOtelSpanAttributes, propagate_attributes from langfuse._client.client import Langfuse from langfuse._client.environment_variables import ( LANGFUSE_FLUSH_AT, @@ -400,3 +402,125 @@ def test_on_end_formats_span_only_when_debug_enabled( assert ("Processing span name='llm-call'" in caplog.text) == bool( expected_formatter_calls ) + + +@pytest.fixture +def skills_tracer(memory_exporter): + providers = [] + + def create(skills, *, span_limits=None): + provider = TracerProvider(span_limits=span_limits) + provider.add_span_processor( + LangfuseSpanProcessor( + public_key="pk-test", + secret_key="sk-test", + base_url="http://localhost:3000", + span_exporter=memory_exporter, + should_export_span=lambda span: True, + skills=skills, + ) + ) + providers.append(provider) + return provider.get_tracer("test-skills"), provider + + yield create + for provider in providers: + provider.shutdown() + + +def test_skills_attachment_default_and_nested_overrides(skills_tracer, memory_exporter): + key = LangfuseOtelSpanAttributes.OBSERVATION_SKILLS_AVAILABLE + catalog = [{"name": "configured", "version": 2}] + tracer, provider = skills_tracer({"available": catalog}) + with tracer.start_as_current_span("default") as span: + assert key not in span.attributes + with propagate_attributes(attach_skills=True): + with tracer.start_as_current_span("enabled") as span: + assert json.loads(span.attributes[key]) == catalog + with propagate_attributes(attach_skills=False): + with tracer.start_as_current_span("disabled") as span: + assert key not in span.attributes + with propagate_attributes(skills_available=[]): + with tracer.start_as_current_span("explicit") as span: + assert span.attributes[key] == "[]" + with tracer.start_as_current_span("restored"): + pass + provider.force_flush() + default, enabled, disabled, explicit, restored = ( + memory_exporter.get_finished_spans() + ) + assert key not in default.attributes + assert json.loads(enabled.attributes[key]) == catalog + assert key not in disabled.attributes + assert explicit.attributes[key] == "[]" + assert json.loads(restored.attributes[key]) == catalog + + +@pytest.mark.parametrize("propagated", [[], [{"name": "propagated"}]]) +def test_skill_attachment_precedence(skills_tracer, propagated): + key = LangfuseOtelSpanAttributes.OBSERVATION_SKILLS_AVAILABLE + tracer, _ = skills_tracer({"available": [{"name": "configured"}]}) + with propagate_attributes(attach_skills=True, skills_available=propagated): + with tracer.start_as_current_span("propagated") as span: + assert json.loads(span.attributes[key]) == propagated + with tracer.start_as_current_span("direct", attributes={key: "[]"}) as span: + assert span.attributes[key] == "[]" + + +def test_configured_skills_are_serialized_once(skills_tracer, memory_exporter): + skills = [{"name": "configured", "future_field": {"enabled": True}}] + tracer, provider = skills_tracer({"available": skills}) + skills.clear() + with propagate_attributes(attach_skills=True): + span = tracer.start_span("configured") + span.end() + provider.force_flush() + span = memory_exporter.get_finished_spans()[0] + assert json.loads( + span.attributes[LangfuseOtelSpanAttributes.OBSERVATION_SKILLS_AVAILABLE] + ) == [{"name": "configured", "future_field": {"enabled": True}}] + + +def test_configured_skills_respect_span_attribute_limit( + skills_tracer, memory_exporter, caplog +): + tracer, provider = skills_tracer( + {"available": [{"name": "configured"}]}, + span_limits=SpanLimits(max_attributes=1), + ) + with propagate_attributes(attach_skills=True): + span = tracer.start_span("full", attributes={"keep": "value"}) + span.end() + provider.force_flush() + assert dict(memory_exporter.get_finished_spans()[0].attributes) == {"keep": "value"} + assert LangfuseOtelSpanAttributes.OBSERVATION_SKILLS_AVAILABLE in caplog.text + + +def test_skill_catalog_is_scoped_to_its_project(memory_exporter): + provider = TracerProvider() + for project in ("A", "B"): + provider.add_span_processor( + LangfuseSpanProcessor( + public_key=project, + secret_key="test-secret", + base_url="http://localhost:3000", + span_exporter=memory_exporter, + skills={ + "available": [{"name": project}], + }, + ) + ) + try: + tracer = provider.get_tracer("langfuse-sdk", attributes={"public_key": "B"}) + with ( + propagate_attributes(attach_skills=True), + tracer.start_as_current_span("project-b") as span, + ): + assert ( + span.attributes[LangfuseOtelSpanAttributes.OBSERVATION_SKILLS_AVAILABLE] + == '[{"name":"B"}]' + ) + provider.force_flush() + assert len(memory_exporter.get_finished_spans()) == 1 + finally: + provider.shutdown()