From 26d60e1ea2f89503fe64a6625fc1a2d51fee956c Mon Sep 17 00:00:00 2001 From: Pierre-Jean Morel Date: Mon, 18 May 2026 13:39:01 +0200 Subject: [PATCH 1/5] feat: add prompt pull command --- src/bto_langfuse_cli/commands/prompt.py | 66 +++++++++- .../langfuse/langfuse_service.py | 41 ++++++ .../langfuse/prompt_exporter.py | 123 ++++++++++++++++++ 3 files changed, 224 insertions(+), 6 deletions(-) create mode 100644 src/bto_langfuse_cli/langfuse/prompt_exporter.py diff --git a/src/bto_langfuse_cli/commands/prompt.py b/src/bto_langfuse_cli/commands/prompt.py index 4739e13..092793c 100644 --- a/src/bto_langfuse_cli/commands/prompt.py +++ b/src/bto_langfuse_cli/commands/prompt.py @@ -2,23 +2,25 @@ from __future__ import annotations -from typing import Annotated +from pathlib import Path +from typing import Annotated, Literal import typer from bto_langfuse_cli.langfuse.langfuse_service import LangfuseService from bto_langfuse_cli.langfuse.print_plan import print_plan +from bto_langfuse_cli.langfuse.prompt_exporter import MarkdownPromptExporter prompt_app = typer.Typer(help="Manage Langfuse prompts.") @prompt_app.command("promote") def prompt_promote( - args: Annotated[ - tuple[str, str], - typer.Argument(help="src_label target_label)."), - ], - apply: Annotated[bool, typer.Option(help="Apply the promotion automatically")] = False, + args: Annotated[ + tuple[str, str], + typer.Argument(help="src_label target_label)."), + ], + apply: Annotated[bool, typer.Option(help="Apply the promotion automatically")] = False, ) -> None: """Promote a label to another label. @@ -33,3 +35,55 @@ def prompt_promote( if plan.has_changes(): if apply or typer.confirm(f"Apply plan ?"): langfuse.apply(plan, to_label) + + +@prompt_app.command("pull") +def prompt_pull( + label: Annotated[ + str, + typer.Argument(help="The label to pull prompt from."), + ], + output_dir: Annotated[ + Path, + typer.Option("--output-dir", "-o", help="Directory to save the prompts to.", dir_okay=True, file_okay=False) + ] = Path("data/prompts"), + prompt_type: Annotated[ + Literal["text", "chat"] | None, + typer.Option("--type", "-t", help="The type of prompt to pull or all if not specified."), + ] = None, + prompt_format: Annotated[ + Literal["md"], + typer.Option("--format", "-f", help="The output format.") + ] = "md" +) -> None: + """ + Pull prompt to filesystem from a given label and prompt type. + + Examples: + bto_langfuse-cli prompt pull dev -o data/prompts -t chat -- → pull all chat prompts tagged dev to data/prompts. + """ + langfuse = LangfuseService() + output_dir.mkdir(parents=True, exist_ok=True) + + if prompt_format == "md": + exporter = MarkdownPromptExporter() + else: + typer.echo(f"Unsupported format: {prompt_format}", err=True) + raise typer.Exit(code=1) + + written = 0 + skipped = 0 + + typer.echo(f"Pulling prompts with label '{label}'...") + + for meta, prompt_client in langfuse.get_prompts_with_meta(label=label, prompt_type=prompt_type): + try: + out_path = exporter.export(meta, prompt_client, output_dir) + if out_path: + typer.echo(f"Wrote {out_path}") + written += 1 + except Exception as e: + typer.echo(f"Error exporting {meta.name}: {e}", err=True) + skipped += 1 + + typer.echo(f"Done. Pulled {written} prompts to {output_dir} (Skipped: {skipped})") \ No newline at end of file diff --git a/src/bto_langfuse_cli/langfuse/langfuse_service.py b/src/bto_langfuse_cli/langfuse/langfuse_service.py index 37e75ef..083ab68 100644 --- a/src/bto_langfuse_cli/langfuse/langfuse_service.py +++ b/src/bto_langfuse_cli/langfuse/langfuse_service.py @@ -1,7 +1,11 @@ import logging +from typing import Iterator, Tuple from bto_langfuse_cli.langfuse.promote_plan import Plan from langfuse import get_client, Langfuse +from langfuse.api.prompts.types.prompt_type import PromptType +from langfuse.api.prompts.types.prompt_meta import PromptMeta +from langfuse.model import TextPromptClient, ChatPromptClient logger = logging.getLogger(__name__) @@ -40,6 +44,43 @@ def _get_prompts(self, label: str) -> dict[str, int]: prompt_data = [self._langfuse.get_prompt(p, label=label) for p in prompts_names] return {p.name: p.version for p in prompt_data} + def get_prompts_with_meta( + self, label: str, prompt_type: str | None = None + ) -> Iterator[Tuple[PromptMeta, TextPromptClient | ChatPromptClient]]: + current_page = 1 + limit = 50 + + while True: + response = self._langfuse.api.prompts.list(label=label, limit=limit, page=current_page) + metas = response.data + + if not metas: + break + + for meta in metas: + if prompt_type: + expected_type = PromptType.TEXT if prompt_type == "text" else PromptType.CHAT + if meta.type != expected_type: + continue + elif meta.type not in (PromptType.TEXT, PromptType.CHAT): + continue + + langfuse_prompt_type = "text" if meta.type is PromptType.TEXT else "chat" + try: + prompt_client = self._langfuse.get_prompt( + meta.name, + label=label, + type=langfuse_prompt_type, + cache_ttl_seconds=0, + ) + yield meta, prompt_client + except Exception as e: + logger.error(f"Failed to fetch prompt {meta.name}: {e}") + + if current_page >= response.meta.total_pages: + break + current_page += 1 + def plan(self, src_label, tgt_label) -> Plan: src_prompts = self._get_prompts(src_label) tgt_prompts = self._get_prompts(tgt_label) diff --git a/src/bto_langfuse_cli/langfuse/prompt_exporter.py b/src/bto_langfuse_cli/langfuse/prompt_exporter.py new file mode 100644 index 0000000..d3b2c91 --- /dev/null +++ b/src/bto_langfuse_cli/langfuse/prompt_exporter.py @@ -0,0 +1,123 @@ +from abc import ABC, abstractmethod +from datetime import datetime +from pathlib import Path + +from langfuse.api import PromptType +from langfuse.api.prompts.types.prompt_meta import PromptMeta +from langfuse.model import TextPromptClient, ChatPromptClient + + +class PromptExporter(ABC): + """Interface for prompt exporters.""" + + @abstractmethod + def export( + self, + meta: PromptMeta, + prompt_client: TextPromptClient | ChatPromptClient, + output_dir: Path, + ) -> Path | None: + """ + Export the prompt as a file to the given directory. + + Args: + meta: The metadata of the prompt from the list API. + prompt_client: The detailed prompt client object. + output_dir: The directory where the prompt should be saved. + + Returns: + The path to the generated file, or None if skipped. + """ + pass + + +class MarkdownPromptExporter(PromptExporter): + """ + Exports prompts to Markdown files with YAML frontmatter + and XML section for Chat prompts. + """ + + def _yaml_quote(self, s: str) -> str: + if s == "": + return '""' + if any(c in s for c in (":", "#", "\n", '"')) or s.strip() != s: + escaped = s.replace("\\", "\\\\").replace('"', '\\"') + return f'"{escaped}"' + return s + + def _format_yaml_block_list(self, key: str, items: list[str]) -> str: + if not items: + return f"{key}: []" + lines = [f"{key}:"] + for item in items: + lines.append(f" - {self._yaml_quote(item)}") + return "\n".join(lines) + + def _metadata_header( + self, + name: str, + version: int, + prompt_type: str, + labels: list[str], + tags: list[str], + last_updated_at: datetime, + ) -> str: + lines = [ + "---", + f"name: {self._yaml_quote(name)}", + f"version: {version}", + f"type: {prompt_type}", + self._format_yaml_block_list("labels", labels), + self._format_yaml_block_list("tags", tags), + f"last_updated_at: {self._yaml_quote(last_updated_at.isoformat())}", + "---", + ] + return "\n".join(lines) + "\n" + + def _serialize_chat_prompt(self, client: ChatPromptClient) -> str: + parts: list[str] = [] + for msg in client.prompt: + if msg["type"] == "message": + role = msg["role"] + content = msg["content"] + parts.append(f"<{role}>\n\n{content}\n\n") + elif msg["type"] == "placeholder": + name = msg["name"] + parts.append(f"{{{{{name}}}}}") + else: + raise ValueError(f"Unsupported chat message entry: {msg!r}") + return "\n\n".join(parts) + + def export( + self, + meta: PromptMeta, + prompt_client: TextPromptClient | ChatPromptClient, + output_dir: Path, + ) -> Path | None: + out_path = Path(output_dir, meta.name).with_suffix(".md") + out_path.parent.mkdir(parents=True, exist_ok=True) + + langfuse_prompt_type = "text" if meta.type is PromptType.TEXT else "chat" + + header = self._metadata_header( + name=prompt_client.name, + version=prompt_client.version, + prompt_type=langfuse_prompt_type, + labels=list(prompt_client.labels), + tags=list(prompt_client.tags), + last_updated_at=meta.last_updated_at, + ) + + if meta.type is PromptType.TEXT: + if not isinstance(prompt_client, TextPromptClient): + raise ValueError(f"Expected TextPromptClient for {meta.name}") + body = prompt_client.prompt + else: + if not isinstance(prompt_client, ChatPromptClient): + raise ValueError(f"Expected ChatPromptClient for {meta.name}") + body = self._serialize_chat_prompt(prompt_client) + + combined = f"{header}\n{body}\n" + out_path.write_text(combined, encoding="utf-8") + return out_path + From 0d97fd4d3e8d0001eb7e9ba6242db18a3bf82e52 Mon Sep 17 00:00:00 2001 From: Pierre-Jean Morel Date: Mon, 18 May 2026 13:46:48 +0200 Subject: [PATCH 2/5] docs: updated docs to match new CLI name and new pull command --- README.md | 45 +++++++++++++++++++++++++++++++++++++++------ 1 file changed, 39 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 7772a86..9bd8615 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,4 @@ -# langfuse-cli +# bto-langfuse-cli A set of CLI tools to interact with Langfuse, especially for prompt management. @@ -16,7 +16,7 @@ Or you can install it using pip if you have the package: pip install . ``` -The CLI command is named `langfuse-cli`. +The CLI command is named `bto-langfuse-cli`. ## Configuration @@ -46,7 +46,7 @@ The `promote` command allows you to copy a label from one version of all prompts **Command:** ```bash -langfuse-cli prompt promote [--apply] +bto-langfuse-cli prompt promote [--apply] ``` **Options:** @@ -56,20 +56,53 @@ langfuse-cli prompt promote [--apply] 1. **Dry run / Confirmation mode:** ```bash - langfuse-cli prompt promote dev uat + bto-langfuse-cli prompt promote dev uat ``` This will show a plan of which prompts will be updated (adding the `uat` label to all prompts that currently have the `dev` label) and ask for confirmation before applying. 2. **Auto-apply mode:** ```bash - langfuse-cli prompt promote dev uat --apply + bto-langfuse-cli prompt promote dev uat --apply ``` This will automatically update the labels for all prompts matching the source label. +#### Pull Prompts + +The `pull` command downloads prompts from Langfuse to your local filesystem based on a specific label. + +**Command:** +```bash +bto-langfuse-cli prompt pull