Export langfuse prompts

README.md

Migrate prompts between langfuse instances/projects

Let's assume that you want to migrate prompt from A to B.

1. mkdir -p data. This will contain the exported prompts.
2. Set env var to A's keys
3. Run python migrate.py export-prompts
4. Set env var to B's keys
5. Run python migrate.py import-prompts

migrate.py

import pickle

import async_typer as typer
from langfuse import get_client
from langfuse.api import (
    CreatePromptRequest_Chat,
    CreatePromptRequest_Text,
    Prompt,
    Prompt_Chat,
    Prompt_Text,
)
from tqdm import tqdm
from tqdm.asyncio import tqdm_asyncio as asyncio

langfuse = get_client()
app = typer.AsyncTyper(pretty_exceptions_enable=False)


@app.async_command()
async def export_prompts(output_file: str = "data/prompts.pk", limit: int = 100):
    api = langfuse.async_api.prompts
    all_prompts_meta = (await api.list(limit=limit)).data

    fetch_prompt_task = []

    for prompt in all_prompts_meta:
        for version in prompt.versions:
            fetch_prompt_task.append(api.get(prompt_name=prompt.name, version=version))

    all_prompts: list[Prompt] = await asyncio.gather(
        *fetch_prompt_task, desc="Fetching prompts"
    )

    with open(output_file, "wb") as f:
        pickle.dump(all_prompts, f)

    typer.echo(f"Successfully exported {len(all_prompts)} prompts to {output_file}")


@app.async_command()
async def import_prompts(input_file: str = "data/prompts.pk"):
    api = langfuse.async_api.prompts
    with open(input_file, "rb") as f:
        prompts: list[Prompt] = pickle.load(f)

    for prompt in tqdm(prompts, desc="Importing prompts"):
        match prompt:
            case Prompt_Chat() as chat_prompt:
                req = CreatePromptRequest_Chat(
                    name=chat_prompt.name,
                    prompt=chat_prompt.prompt,
                    config=chat_prompt.config,
                    labels=chat_prompt.labels,
                    tags=chat_prompt.tags,
                    commitMessage=chat_prompt.commit_message,
                    type=chat_prompt.type,
                )
            case Prompt_Text() as text_prompt:
                req = CreatePromptRequest_Text(
                    name=text_prompt.name,
                    prompt=text_prompt.prompt,
                    config=text_prompt.config,
                    labels=text_prompt.labels,
                    tags=text_prompt.tags,
                    commitMessage=text_prompt.commit_message,
                    type=text_prompt.type,
                )
        # Langfuse doesn't allow concurrent creation of prompt under same name
        # Let's just do this without parallelism.
        await api.create(request=req)

    typer.echo(f"Successfully imported {len(prompts)} prompts from {input_file}")


if __name__ == "__main__":
    app()