From 00b46842225ad940c27c1db01fbaa3a9bf0dfda7 Mon Sep 17 00:00:00 2001 From: Robin DE BASTOS Date: Tue, 7 Jul 2026 10:44:33 +0200 Subject: [PATCH 1/2] feat(skill): add skills management client, cache and manager methods Hand-written high-level skills API mirroring the prompt manager: - SkillCache (TTL, background stale-while-revalidate refresh, prefix invalidation) in langfuse/_utils/skill_cache.py - SkillClient wrapper with .compile() variable substitution of the instructions body (single client; skills have no chat/text split) - Langfuse client methods get_skill / create_skill / update_skill / delete_skill / clear_skill_cache with caching, fallback and retries - unit tests for cache TTL and client compilation Depends on the auto-generated langfuse.api skills client (api-spec-bot regeneration from the Fern contract in langfuse/langfuse). --- langfuse/_client/client.py | 301 ++++++++++++++++++++++ langfuse/_client/environment_variables.py | 10 + langfuse/_client/resource_manager.py | 5 + langfuse/_utils/skill_cache.py | 231 +++++++++++++++++ langfuse/model.py | 75 ++++++ tests/unit/test_skill.py | 215 ++++++++++++++++ 6 files changed, 837 insertions(+) create mode 100644 langfuse/_utils/skill_cache.py create mode 100644 tests/unit/test_skill.py diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index 6ce72d27a..081fc3974 100644 --- a/langfuse/_client/client.py +++ b/langfuse/_client/client.py @@ -94,10 +94,12 @@ from langfuse._utils.environment import get_common_release_envs from langfuse._utils.parse_error import handle_fern_exception from langfuse._utils.prompt_cache import PromptCache +from langfuse._utils.skill_cache import SkillCache from langfuse.api import ( AsyncLangfuseAPI, CreateChatPromptRequest, CreateChatPromptType, + CreateSkillRequest, CreateTextPromptRequest, Dataset, DatasetItem, @@ -113,6 +115,7 @@ Prompt_Chat, Prompt_Text, ScoreBody, + Skill, TraceBody, ) from langfuse.batch_evaluation import ( @@ -141,6 +144,7 @@ ChatMessageWithPlaceholdersDict, ChatPromptClient, PromptClient, + SkillClient, TextPromptClient, ) from langfuse.types import ( @@ -4055,3 +4059,300 @@ def clear_prompt_cache(self) -> None: """ if self._resources is not None: self._resources.prompt_cache.clear() + + def get_skill( + self, + name: str, + *, + version: Optional[int] = None, + label: Optional[str] = None, + cache_ttl_seconds: Optional[int] = None, + fallback: Optional[str] = None, + max_retries: Optional[int] = None, + fetch_timeout_seconds: Optional[int] = None, + ) -> SkillClient: + """Get a skill. + + This method attempts to fetch the requested skill from the local cache. If the skill is not found + in the cache or if the cached skill has expired, it will try to fetch the skill from the server again + and update the cache. If fetching the new skill fails, and there is an expired skill in the cache, it will + return the expired skill as a fallback. + + Args: + name (str): The name of the skill to retrieve. + + Keyword Args: + version (Optional[int]): The version of the skill to retrieve. If no label and version is specified, the `production` label is returned. Specify either version or label, not both. + label: Optional[str]: The label of the skill to retrieve. If no label and version is specified, the `production` label is returned. Specify either version or label, not both. + cache_ttl_seconds: Optional[int]: Time-to-live in seconds for caching the skill. Must be specified as a + keyword argument. If not set, defaults to 60 seconds. Disables caching if set to 0. + fallback: Optional[str]: The skill instructions to return if fetching the skill fails. Important on the first call where no cached skill is available. Follows Langfuse skill formatting with double curly braces for variables. Defaults to None. + max_retries: Optional[int]: The maximum number of retries in case of API/network errors. Defaults to 2. The maximum value is 4. Retries have an exponential backoff with a maximum delay of 10 seconds. + fetch_timeout_seconds: Optional[int]: The timeout in milliseconds for fetching the skill. Defaults to the default timeout set on the SDK, which is 5 seconds per default. + + Returns: + The skill object retrieved from the cache or directly fetched if not cached or expired, as a SkillClient. + + Raises: + Exception: Propagates any exceptions raised during the fetching of a new skill, unless there is an + expired skill in the cache, in which case it logs a warning and returns the expired skill. + """ + if self._resources is None: + raise Error( + "SDK is not correctly initialized. Check the init logs for more details." + ) + if version is not None and label is not None: + raise ValueError("Cannot specify both version and label at the same time.") + + if not name: + raise ValueError("Skill name cannot be empty.") + + cache_key = SkillCache.generate_cache_key(name, version=version, label=label) + bounded_max_retries = self._get_bounded_max_retries( + max_retries, default_max_retries=2, max_retries_upper_bound=4 + ) + + langfuse_logger.debug(f"Getting skill '{cache_key}'") + cached_skill = self._resources.skill_cache.get(cache_key) + + if cached_skill is None or cache_ttl_seconds == 0: + langfuse_logger.debug( + f"Skill '{cache_key}' not found in cache or caching disabled." + ) + try: + return self._fetch_skill_and_update_cache( + name, + version=version, + label=label, + ttl_seconds=cache_ttl_seconds, + max_retries=bounded_max_retries, + fetch_timeout_seconds=fetch_timeout_seconds, + ) + except Exception as e: + if fallback: + langfuse_logger.warning( + f"Returning fallback skill for '{cache_key}' due to fetch error: {e}" + ) + + return SkillClient( + skill=Skill( + name=name, + version=version or 0, + description=None, + instructions=fallback, + metadata=None, + allowed_tools=[], + labels=[label] if label else [], + tags=[], + commit_message=None, + ), + is_fallback=True, + ) + + raise e + + if cached_skill.is_expired(): + langfuse_logger.debug(f"Stale skill '{cache_key}' found in cache.") + try: + # refresh skill in background thread, refresh_skill deduplicates tasks + langfuse_logger.debug(f"Refreshing skill '{cache_key}' in background.") + + def refresh_task() -> None: + self._fetch_skill_and_update_cache( + name, + version=version, + label=label, + ttl_seconds=cache_ttl_seconds, + max_retries=bounded_max_retries, + fetch_timeout_seconds=fetch_timeout_seconds, + ) + + self._resources.skill_cache.add_refresh_skill_task_if_current( + cache_key, + cached_skill, + refresh_task, + ) + langfuse_logger.debug( + f"Returning stale skill '{cache_key}' from cache." + ) + # return stale skill + return cached_skill.value + + except Exception as e: + langfuse_logger.warning( + f"Error when refreshing cached skill '{cache_key}', returning cached version. Error: {e}" + ) + # creation of refresh skill task failed, return stale skill + return cached_skill.value + + return cached_skill.value + + def _fetch_skill_and_update_cache( + self, + name: str, + *, + version: Optional[int] = None, + label: Optional[str] = None, + ttl_seconds: Optional[int] = None, + max_retries: int, + fetch_timeout_seconds: Optional[int], + ) -> SkillClient: + cache_key = SkillCache.generate_cache_key(name, version=version, label=label) + langfuse_logger.debug(f"Fetching skill '{cache_key}' from server...") + + try: + + @backoff.on_exception( + backoff.constant, Exception, max_tries=max_retries + 1, logger=None + ) + def fetch_skills() -> Any: + return self.api.skills.get( + self._url_encode(name), + version=version, + label=label, + request_options={ + "timeout_in_seconds": fetch_timeout_seconds, + } + if fetch_timeout_seconds is not None + else None, + ) + + skill_response = fetch_skills() + + skill = SkillClient(skill_response) + + if self._resources is not None: + self._resources.skill_cache.set(cache_key, skill, ttl_seconds) + + return skill + + except NotFoundError as not_found_error: + langfuse_logger.warning( + f"Skill '{cache_key}' not found during refresh, evicting from cache." + ) + if self._resources is not None: + self._resources.skill_cache.delete(cache_key) + raise not_found_error + + except Exception as e: + langfuse_logger.error(f"Error while fetching skill '{cache_key}': {str(e)}") + raise e + + def create_skill( + self, + *, + name: str, + instructions: str, + description: Optional[str] = None, + labels: List[str] = [], + tags: Optional[List[str]] = None, + metadata: Optional[Any] = None, + allowed_tools: Optional[List[str]] = None, + commit_message: Optional[str] = None, + ) -> SkillClient: + """Create a new skill in Langfuse. + + Keyword Args: + name : The name of the skill to be created. + instructions : The instructions content of the skill to be created. May contain {{variable}} placeholders. + description: Optional human-readable description of the skill. Defaults to None. + labels: The labels of the skill. Defaults to None. To create a default-served skill, add the 'production' label. + tags: The tags of the skill. Defaults to None. Will be applied to all versions of the skill. + metadata: Additional structured data to be saved with the skill. Defaults to None. + allowed_tools: The list of tool names the skill is allowed to use. Defaults to None. + commit_message: Optional string describing the change. + + Returns: + SkillClient: The created skill. + """ + try: + langfuse_logger.debug(f"Creating skill {name=}, {labels=}") + + request = CreateSkillRequest( + name=name, + instructions=instructions, + description=description, + labels=labels, + tags=tags, + metadata=metadata, + allowed_tools=allowed_tools, + commit_message=commit_message, + ) + + server_skill = self.api.skills.create(request=request) + + if self._resources is not None: + self._resources.skill_cache.invalidate(name) + + return SkillClient(skill=cast(Skill, server_skill)) + + except Error as e: + handle_fern_exception(e) + raise e + + def update_skill( + self, + *, + name: str, + version: int, + new_labels: List[str] = [], + ) -> Any: + """Update an existing skill version in Langfuse. The Langfuse SDK skill cache is invalidated for all skills with the specified name. + + Args: + name (str): The name of the skill to update. + version (int): The version number of the skill to update. + new_labels (List[str], optional): New labels to assign to the skill version. Labels are unique across versions. The "latest" label is reserved and managed by Langfuse. Defaults to []. + + Returns: + Skill: The updated skill from the Langfuse API. + + """ + updated_skill = self.api.skill_version.update( + name=self._url_encode(name), + version=version, + new_labels=new_labels, + ) + + if self._resources is not None: + self._resources.skill_cache.invalidate(name) + + return updated_skill + + def delete_skill( + self, + name: str, + *, + version: Optional[int] = None, + label: Optional[str] = None, + ) -> Any: + """Delete a skill in Langfuse. The Langfuse SDK skill cache is invalidated for all skills with the specified name. + + Args: + name (str): The name of the skill to delete. + + Keyword Args: + version (Optional[int]): The version of the skill to delete. If neither version nor label is specified, all versions are deleted. + label (Optional[str]): The label of the skill version to delete. + + """ + result = self.api.skills.delete( + self._url_encode(name), + version=version, + label=label, + ) + + if self._resources is not None: + self._resources.skill_cache.invalidate(name) + + return result + + def clear_skill_cache(self) -> None: + """Clear the entire skill cache, removing all cached skills. + + This method is useful when you want to force a complete refresh of all + cached skills, for example after major updates or when you need to + ensure the latest versions are fetched from the server. + """ + if self._resources is not None: + self._resources.skill_cache.clear() diff --git a/langfuse/_client/environment_variables.py b/langfuse/_client/environment_variables.py index 6b421578a..aa92ae7f9 100644 --- a/langfuse/_client/environment_variables.py +++ b/langfuse/_client/environment_variables.py @@ -156,3 +156,13 @@ **Default value**: ``60`` """ + +LANGFUSE_SKILL_CACHE_DEFAULT_TTL_SECONDS = "LANGFUSE_SKILL_CACHE_DEFAULT_TTL_SECONDS" +""" +.. envvar: LANGFUSE_SKILL_CACHE_DEFAULT_TTL_SECONDS + +Controls the default time-to-live (TTL) in seconds for cached skills. +This setting determines how long skill responses are cached before they expire. + +**Default value**: ``60`` +""" diff --git a/langfuse/_client/resource_manager.py b/langfuse/_client/resource_manager.py index 67c44920a..562dc8110 100644 --- a/langfuse/_client/resource_manager.py +++ b/langfuse/_client/resource_manager.py @@ -47,6 +47,7 @@ from langfuse._utils.environment import get_common_release_envs from langfuse._utils.prompt_cache import PromptCache from langfuse._utils.request import LangfuseClient +from langfuse._utils.skill_cache import SkillCache from langfuse.api import AsyncLangfuseAPI, LangfuseAPI from langfuse.logger import langfuse_logger from langfuse.types import MaskFunction, MaskOtelSpansFunction @@ -275,6 +276,9 @@ def _initialize_instance( # Prompt cache self.prompt_cache = PromptCache() + # Skill cache + self.skill_cache = SkillCache() + # Register shutdown handler atexit.register(self.shutdown) @@ -463,6 +467,7 @@ def _at_fork_reinit(self) -> None: self._init_media_manager() self._init_consumer_threads() self.prompt_cache = PromptCache() + self.skill_cache = SkillCache() except Exception as e: langfuse_logger.error( f"[PID {os.getpid()}] Failed to reinitialize consumer threads after fork: {e}. " diff --git a/langfuse/_utils/skill_cache.py b/langfuse/_utils/skill_cache.py new file mode 100644 index 000000000..3c1c5951e --- /dev/null +++ b/langfuse/_utils/skill_cache.py @@ -0,0 +1,231 @@ +"""@private""" + +import atexit +import os +from datetime import datetime +from queue import Queue +from threading import RLock, Thread +from typing import Callable, Dict, List, Optional, Set + +from langfuse._client.environment_variables import ( + LANGFUSE_SKILL_CACHE_DEFAULT_TTL_SECONDS, +) +from langfuse.logger import langfuse_logger as logger +from langfuse.model import SkillClient + +DEFAULT_SKILL_CACHE_TTL_SECONDS = int( + os.getenv(LANGFUSE_SKILL_CACHE_DEFAULT_TTL_SECONDS, 60) +) + +DEFAULT_SKILL_CACHE_REFRESH_WORKERS = 1 +_SHUTDOWN_SENTINEL = object() + + +class SkillCacheItem: + def __init__(self, skill: SkillClient, ttl_seconds: int): + self.value = skill + self._expiry = ttl_seconds + self.get_epoch_seconds() + + def is_expired(self) -> bool: + return self.get_epoch_seconds() > self._expiry + + @staticmethod + def get_epoch_seconds() -> int: + return int(datetime.now().timestamp()) + + +class SkillCacheRefreshConsumer(Thread): + _queue: Queue + _identifier: int + running: bool = True + + def __init__(self, queue: Queue, identifier: int): + super().__init__() + self.daemon = True + self._queue = queue + self._identifier = identifier + + def run(self) -> None: + while self.running: + task = self._queue.get() + + if task is _SHUTDOWN_SENTINEL: + self._queue.task_done() + break + + logger.debug( + f"SkillCacheRefreshConsumer processing task, {self._identifier}" + ) + try: + task() + # Task failed, but we still consider it processed + except Exception as e: + logger.warning( + f"SkillCacheRefreshConsumer encountered an error, cache was not refreshed: {self._identifier}, {e}" + ) + + self._queue.task_done() + + def pause(self) -> None: + """Pause the consumer.""" + self.running = False + + +class SkillCacheTaskManager(object): + _consumers: List[SkillCacheRefreshConsumer] + _threads: int + _queue: Queue + _processing_keys: Set[str] + _lock: RLock + + def __init__(self, threads: int = 1): + self._queue = Queue() + self._consumers = [] + self._threads = threads + self._processing_keys = set() + self._lock = RLock() + + for i in range(self._threads): + consumer = SkillCacheRefreshConsumer(self._queue, i) + consumer.start() + self._consumers.append(consumer) + + atexit.register(self.shutdown) + + def add_task(self, key: str, task: Callable[[], None]) -> None: + with self._lock: + if key not in self._processing_keys: + logger.debug(f"Adding skill cache refresh task for key: {key}") + self._processing_keys.add(key) + wrapped_task = self._wrap_task(key, task) + self._queue.put((wrapped_task)) + else: + logger.debug( + f"Skill cache refresh task already submitted for key: {key}" + ) + + def active_tasks(self) -> int: + with self._lock: + return len(self._processing_keys) + + def wait_for_idle(self) -> None: + self._queue.join() + + def _wrap_task(self, key: str, task: Callable[[], None]) -> Callable[[], None]: + def wrapped() -> None: + logger.debug(f"Refreshing skill cache for key: {key}") + try: + task() + finally: + with self._lock: + self._processing_keys.remove(key) + logger.debug(f"Refreshed skill cache for key: {key}") + + return wrapped + + def shutdown(self) -> None: + logger.debug( + f"Shutting down skill refresh task manager, {len(self._consumers)} consumers,..." + ) + + atexit.unregister(self.shutdown) + + for consumer in self._consumers: + consumer.pause() + + for _ in self._consumers: + self._queue.put(_SHUTDOWN_SENTINEL) + + for consumer in self._consumers: + try: + consumer.join() + except RuntimeError: + # consumer thread has not started + pass + + logger.debug("Shutdown of skill refresh task manager completed.") + + +class SkillCache: + _cache: Dict[str, SkillCacheItem] + _lock: RLock + + _task_manager: SkillCacheTaskManager + """Task manager for refreshing cache""" + + def __init__( + self, max_skill_refresh_workers: int = DEFAULT_SKILL_CACHE_REFRESH_WORKERS + ): + self._cache = {} + self._lock = RLock() + self._task_manager = SkillCacheTaskManager(threads=max_skill_refresh_workers) + logger.debug("Skill cache initialized.") + + def get(self, key: str) -> Optional[SkillCacheItem]: + with self._lock: + return self._cache.get(key, None) + + def set(self, key: str, value: SkillClient, ttl_seconds: Optional[int]) -> None: + if ttl_seconds is None: + ttl_seconds = DEFAULT_SKILL_CACHE_TTL_SECONDS + + with self._lock: + self._cache[key] = SkillCacheItem(value, ttl_seconds) + + def delete(self, key: str) -> None: + with self._lock: + self._cache.pop(key, None) + + def invalidate(self, skill_name: str) -> None: + """Invalidate all cached skills with the given skill name.""" + with self._lock: + for key in list(self._cache): + if key.startswith(skill_name): + del self._cache[key] + + def add_refresh_skill_task(self, key: str, fetch_func: Callable[[], None]) -> None: + logger.debug(f"Submitting refresh task for key: {key}") + self._task_manager.add_task(key, fetch_func) + + def add_refresh_skill_task_if_current( + self, + key: str, + expected_item: SkillCacheItem, + fetch_func: Callable[[], None], + ) -> None: + with self._lock: + current_item = self._cache.get(key) + if ( + current_item is not None + and current_item is not expected_item + and not current_item.is_expired() + ): + logger.debug( + f"Skipping refresh task for key: {key} because cache is already fresh." + ) + return + + self.add_refresh_skill_task(key, fetch_func) + + def clear(self) -> None: + """Clear the entire skill cache, removing all cached skills.""" + with self._lock: + self._cache.clear() + + @staticmethod + def generate_cache_key( + name: str, *, version: Optional[int], label: Optional[str] + ) -> str: + parts = [name] + + if version is not None: + parts.append(f"version:{version}") + + elif label is not None: + parts.append(f"label:{label}") + + else: + # Default to production labeled skill + parts.append("label:production") + + return "-".join(parts) diff --git a/langfuse/model.py b/langfuse/model.py index 69d721597..2f0dbde19 100644 --- a/langfuse/model.py +++ b/langfuse/model.py @@ -16,6 +16,7 @@ Prompt, Prompt_Chat, Prompt_Text, + Skill, TraceWithFullDetails, # noqa ) from langfuse.logger import langfuse_logger @@ -475,3 +476,77 @@ def get_langchain_prompt( PromptClient = Union[TextPromptClient, ChatPromptClient] + + +class SkillClient: + """High-level wrapper around a Langfuse ``Skill``. + + Unlike prompts, skills do not have a chat/text split. A skill wraps a single + set of ``instructions`` that may contain ``{{variable}}`` placeholders, which + can be substituted via :meth:`compile`. + """ + + name: str + version: int + description: Optional[str] + instructions: str + metadata: Optional[Any] + allowed_tools: List[str] + labels: List[str] + tags: List[str] + commit_message: Optional[str] + + def __init__(self, skill: Skill, is_fallback: bool = False): + self.name = skill.name + self.version = skill.version + self.description = skill.description + self.instructions = skill.instructions + self.metadata = skill.metadata + self.allowed_tools = skill.allowed_tools or [] + self.labels = skill.labels or [] + self.tags = skill.tags or [] + self.commit_message = skill.commit_message + self.is_fallback = is_fallback + + def compile(self, **kwargs: Union[str, Any]) -> str: + """Compile the skill instructions, substituting any ``{{variable}}`` placeholders. + + Args: + **kwargs: Values for the variables to substitute in the instructions. + + Returns: + The compiled instructions string with all matching variables substituted. + """ + return TemplateParser.compile_template(self.instructions, kwargs) + + @property + def variables(self) -> List[str]: + """Return all the variable names in the skill instructions template.""" + return TemplateParser.find_variable_names(self.instructions) + + def __eq__(self, other: object) -> bool: + if isinstance(other, self.__class__): + return ( + self.name == other.name + and self.version == other.version + and self.instructions == other.instructions + and self.description == other.description + and self.metadata == other.metadata + and self.allowed_tools == other.allowed_tools + ) + + return False + + def dict(self) -> Dict[str, Any]: + """Return a serializable dictionary representation of the skill.""" + return { + "name": self.name, + "version": self.version, + "description": self.description, + "instructions": self.instructions, + "metadata": self.metadata, + "allowed_tools": self.allowed_tools, + "labels": self.labels, + "tags": self.tags, + "commit_message": self.commit_message, + } diff --git a/tests/unit/test_skill.py b/tests/unit/test_skill.py new file mode 100644 index 000000000..16a317a22 --- /dev/null +++ b/tests/unit/test_skill.py @@ -0,0 +1,215 @@ +from types import SimpleNamespace +from unittest.mock import patch + +from langfuse._utils.skill_cache import ( + DEFAULT_SKILL_CACHE_TTL_SECONDS, + SkillCache, + SkillCacheItem, +) +from langfuse.model import SkillClient + + +def _make_skill( + *, + name="my-skill", + version=1, + description="A test skill", + instructions="Help the user with {{task}}.", + metadata=None, + allowed_tools=None, + labels=None, + tags=None, + commit_message=None, +): + """Build a lightweight stand-in for the generated ``Skill`` model. + + ``SkillClient`` only reads attributes off the passed object, so a + ``SimpleNamespace`` avoids depending on the auto-generated ``Skill`` + constructor for these unit tests. + """ + return SimpleNamespace( + name=name, + version=version, + description=description, + instructions=instructions, + metadata=metadata, + allowed_tools=allowed_tools if allowed_tools is not None else [], + labels=labels if labels is not None else [], + tags=tags if tags is not None else [], + commit_message=commit_message, + ) + + +# --------------------------------------------------------------------------- +# SkillClient.compile +# --------------------------------------------------------------------------- + + +def test_compile_substitutes_variables(): + client = SkillClient(_make_skill(instructions="Help the user with {{task}}.")) + + assert client.compile(task="coding") == "Help the user with coding." + + +def test_compile_multiple_variables(): + client = SkillClient( + _make_skill(instructions="{{greeting}}, complete the {{task}}.") + ) + + assert client.compile(greeting="Hi", task="report") == "Hi, complete the report." + + +def test_compile_missing_variable_left_untouched(): + client = SkillClient(_make_skill(instructions="Help with {{task}}.")) + + assert client.compile() == "Help with {{task}}." + + +def test_compile_none_variable_becomes_empty_string(): + client = SkillClient(_make_skill(instructions="Help with {{task}}.")) + + assert client.compile(task=None) == "Help with ." + + +def test_variables_property(): + client = SkillClient(_make_skill(instructions="{{greeting}} {{task}} {{greeting}}")) + + assert client.variables == ["greeting", "task", "greeting"] + + +def test_client_exposes_skill_fields(): + client = SkillClient( + _make_skill( + name="planner", + version=3, + description="Plans things", + metadata={"team": "core"}, + allowed_tools=["search", "write"], + labels=["production"], + tags=["planning"], + commit_message="init", + ) + ) + + assert client.name == "planner" + assert client.version == 3 + assert client.description == "Plans things" + assert client.metadata == {"team": "core"} + assert client.allowed_tools == ["search", "write"] + assert client.labels == ["production"] + assert client.tags == ["planning"] + assert client.commit_message == "init" + assert client.is_fallback is False + + +def test_client_dict_serialization(): + client = SkillClient(_make_skill(name="s", version=2)) + data = client.dict() + + assert data["name"] == "s" + assert data["version"] == 2 + assert data["instructions"] == "Help the user with {{task}}." + assert set(data.keys()) == { + "name", + "version", + "description", + "instructions", + "metadata", + "allowed_tools", + "labels", + "tags", + "commit_message", + } + + +def test_client_equality(): + a = SkillClient(_make_skill()) + b = SkillClient(_make_skill()) + c = SkillClient(_make_skill(instructions="different")) + + assert a == b + assert a != c + assert a != "not-a-skill" + + +def test_none_list_fields_default_to_empty(): + client = SkillClient(_make_skill(allowed_tools=None, labels=None, tags=None)) + + assert client.allowed_tools == [] + assert client.labels == [] + assert client.tags == [] + + +# --------------------------------------------------------------------------- +# SkillCache TTL / expiry +# --------------------------------------------------------------------------- + + +@patch.object(SkillCacheItem, "get_epoch_seconds") +def test_cache_item_expiry(mock_time): + mock_time.return_value = 0 + item = SkillCacheItem(SkillClient(_make_skill()), ttl_seconds=60) + + assert item.is_expired() is False + + mock_time.return_value = 59 + assert item.is_expired() is False + + mock_time.return_value = 61 + assert item.is_expired() is True + + +@patch.object(SkillCacheItem, "get_epoch_seconds") +def test_cache_set_get_and_default_ttl(mock_time): + mock_time.return_value = 0 + cache = SkillCache() + value = SkillClient(_make_skill()) + + cache.set("my-skill", value, ttl_seconds=None) + cached = cache.get("my-skill") + + assert cached is not None + assert cached.value is value + assert cached.is_expired() is False + + mock_time.return_value = DEFAULT_SKILL_CACHE_TTL_SECONDS + 1 + assert cache.get("my-skill").is_expired() is True + + +def test_cache_delete_and_clear(): + cache = SkillCache() + cache.set("a", SkillClient(_make_skill(name="a")), ttl_seconds=60) + cache.set("b", SkillClient(_make_skill(name="b")), ttl_seconds=60) + + cache.delete("a") + assert cache.get("a") is None + assert cache.get("b") is not None + + cache.clear() + assert cache.get("b") is None + + +def test_cache_invalidate_by_name_prefix(): + cache = SkillCache() + cache.set("my-skill-version:1", SkillClient(_make_skill()), ttl_seconds=60) + cache.set("my-skill-label:production", SkillClient(_make_skill()), ttl_seconds=60) + cache.set("other-skill", SkillClient(_make_skill(name="other")), ttl_seconds=60) + + cache.invalidate("my-skill") + + assert cache.get("my-skill-version:1") is None + assert cache.get("my-skill-label:production") is None + assert cache.get("other-skill") is not None + + +def test_generate_cache_key(): + assert SkillCache.generate_cache_key("s", version=2, label=None) == "s-version:2" + assert ( + SkillCache.generate_cache_key("s", version=None, label="staging") + == "s-label:staging" + ) + # Defaults to the production label when neither version nor label given + assert ( + SkillCache.generate_cache_key("s", version=None, label=None) + == "s-label:production" + ) From 1952f274e33e02075933be8705d0222f2705e962 Mon Sep 17 00:00:00 2001 From: Robin DE BASTOS Date: Tue, 7 Jul 2026 14:32:46 +0200 Subject: [PATCH 2/2] =?UTF-8?q?fix(skill):=20address=20review=20=E2=80=94?= =?UTF-8?q?=20mutable=20defaults,=20cache=20prefix=20anchor,=20docstring?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - create_skill/update_skill: default labels/new_labels to None and coerce to [] in the body (avoid shared mutable default list) - SkillCache.invalidate: anchor prefix match with '-' separator so a skill named 'foo' cannot evict 'foobar-*' cache entries - get_skill: docstring fetch_timeout_seconds is seconds, not milliseconds --- langfuse/_client/client.py | 10 +++++++--- langfuse/_utils/skill_cache.py | 3 ++- 2 files changed, 9 insertions(+), 4 deletions(-) diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index 081fc3974..f1158a14c 100644 --- a/langfuse/_client/client.py +++ b/langfuse/_client/client.py @@ -4088,7 +4088,7 @@ def get_skill( keyword argument. If not set, defaults to 60 seconds. Disables caching if set to 0. fallback: Optional[str]: The skill instructions to return if fetching the skill fails. Important on the first call where no cached skill is available. Follows Langfuse skill formatting with double curly braces for variables. Defaults to None. max_retries: Optional[int]: The maximum number of retries in case of API/network errors. Defaults to 2. The maximum value is 4. Retries have an exponential backoff with a maximum delay of 10 seconds. - fetch_timeout_seconds: Optional[int]: The timeout in milliseconds for fetching the skill. Defaults to the default timeout set on the SDK, which is 5 seconds per default. + fetch_timeout_seconds: Optional[int]: The timeout in seconds for fetching the skill. Defaults to the default timeout set on the SDK, which is 5 seconds per default. Returns: The skill object retrieved from the cache or directly fetched if not cached or expired, as a SkillClient. @@ -4244,7 +4244,7 @@ def create_skill( name: str, instructions: str, description: Optional[str] = None, - labels: List[str] = [], + labels: Optional[List[str]] = None, tags: Optional[List[str]] = None, metadata: Optional[Any] = None, allowed_tools: Optional[List[str]] = None, @@ -4265,6 +4265,8 @@ def create_skill( Returns: SkillClient: The created skill. """ + labels = labels or [] + try: langfuse_logger.debug(f"Creating skill {name=}, {labels=}") @@ -4295,7 +4297,7 @@ def update_skill( *, name: str, version: int, - new_labels: List[str] = [], + new_labels: Optional[List[str]] = None, ) -> Any: """Update an existing skill version in Langfuse. The Langfuse SDK skill cache is invalidated for all skills with the specified name. @@ -4308,6 +4310,8 @@ def update_skill( Skill: The updated skill from the Langfuse API. """ + new_labels = new_labels or [] + updated_skill = self.api.skill_version.update( name=self._url_encode(name), version=version, diff --git a/langfuse/_utils/skill_cache.py b/langfuse/_utils/skill_cache.py index 3c1c5951e..1436e5a6d 100644 --- a/langfuse/_utils/skill_cache.py +++ b/langfuse/_utils/skill_cache.py @@ -178,9 +178,10 @@ def delete(self, key: str) -> None: def invalidate(self, skill_name: str) -> None: """Invalidate all cached skills with the given skill name.""" + prefix = skill_name + "-" with self._lock: for key in list(self._cache): - if key.startswith(skill_name): + if key.startswith(prefix): del self._cache[key] def add_refresh_skill_task(self, key: str, fetch_func: Callable[[], None]) -> None: