文档 / Langfuse 提示词的 GitHub 集成

Langfuse 提示词的 GitHub 集成

有两种方法将 Langfuse 提示词与 GitHub 集成:

GitHub Repository Dispatch - 当提示词变更时触发 CI/CD 工作流。这不需要额外基础设施。
将 Langfuse 提示词同步到仓库 - 将提示词存储在仓库中的特定文件里。这涉及一个 webhook 服务器,监听提示词版本变更并将它们提交到仓库。

触发 GitHub Actions

当 Langfuse 提示词变更时,使用 repository_dispatch 事件触发 GitHub Actions 工作流。

mermaid

sequenceDiagram
    participant User as User/Team
    participant LF as Langfuse
    participant GH as GitHub API
    participant Actions as GitHub Actions

    User->>LF: Update prompt in Langfuse
    LF->>GH: POST /repos/owner/repo/dispatches
    GH->>Actions: Trigger repository_dispatch event
    Actions->>Actions: Run CI workflow (tests, deploy, etc.)
    Note over User,Actions: Prompt changes trigger automated workflows

1. 创建 GitHub Workflow

.github/workflows/langfuse-ci.yml:

yaml

name: Langfuse Prompt CI
on:
  repository_dispatch:
    types: [langfuse-prompt-update]
  workflow_dispatch:

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Run tests
        run: |
          echo "Testing prompt: ${{ github.event.client_payload.prompt.name }} v${{ github.event.client_payload.prompt.version }}"
          # Add your test commands
          # npm test
          # python -m pytest

  deploy:
    needs: test
    runs-on: ubuntu-latest
    if: contains(github.event.client_payload.prompt.labels, 'production')
    steps:
      - uses: actions/checkout@v4
      - name: Deploy to production
        run: |
          echo "Deploying ${{ github.event.client_payload.prompt.name }} v${{ github.event.client_payload.prompt.version }}"
          # Your deployment commands

访问 webhook 数据: 使用 github.event.client_payload.* 访问提示词数据:

yaml

# Example: Access webhook data in your workflow
- name: Process prompt data
  run: |
    echo "Action: ${{ github.event.client_payload.action }}"
    echo "Prompt: ${{ github.event.client_payload.prompt.name }}"
    echo "Version: ${{ github.event.client_payload.prompt.version }}"
    echo "Labels: ${{ github.event.client_payload.prompt.labels }}"

- name: Deploy only production prompts
  if: contains(github.event.client_payload.prompt.labels, 'production')
  run: echo "Deploying production prompt"

2. 为 Actions 创建 GitHub Token

步骤:

GitHub Settings > Developer settings > Personal access tokens
Generate new token(classic 或 fine-grained)
选择 scope(见下表)

Token 类型	所需权限
Personal Access Token (classic)	`repo` scope(公开仓库)或 `public_repo` scope(私有仓库)
Fine-grained PAT 或 GitHub App	对 `actions` 的 `read` 和 `write`

3. 在 Langfuse 中配置 GitHub Action

在你的 Langfuse 项目中进入 Prompts > Automations。
点击 Create Automation。
选择 GitHub Repository Dispatch。
配置自动化:
- Dispatch URL:https://api.github.com/repos///dispatches(将 和 替换为你的值)
- Event Type:langfuse-prompt-update(必须与你的 GitHub workflow 中的 type 匹配)
- GitHub Token:输入你的 GitHub Personal Access Token。它将被安全存储。

4. 测试 GitHub Actions 集成

在 Langfuse 中更新一个带 production 标签的提示词
检查 GitHub Actions 标签页中触发的工作流
验证 test 和 deploy 作业都成功运行

将 Langfuse 提示词同步到仓库

使用提示词版本 Webhooks自动将提示词变更从 Langfuse 同步到 GitHub。这为提示词启用了版本控制,并可以触发 CI/CD 工作流。

同步工作流概览

每当你在 Langfuse 中保存新的提示词版本时,它会自动提交到你的 GitHub 仓库。通过此设置,你还可以在提示词变更时触发 CI/CD 工作流。

mermaid

sequenceDiagram
    participant User as User/Team
    participant LF as Langfuse
    participant FastAPI as FastAPI Server
    participant GitHub as GitHub

    User->>LF: Set up webhooks
    User->>LF: Modify a prompt
    LF->>FastAPI: POST /webhook/prompt (JSON payload)
    FastAPI->>GitHub: GET file SHA (if exists)
    GitHub-->>FastAPI: Return current file SHA
    FastAPI->>GitHub: PUT /repos/:owner/:repo/contents/:path
    GitHub->>GitHub: Create/update commit with prompt
    GitHub-->>FastAPI: ✅ Commit successful
    FastAPI-->>LF: 201 Created response
    Note over User,GitHub: Prompt changes now version-controlled in GitHub

同步的前提条件

Langfuse 项目: 具有项目所有者访问权限的提示词设置
GitHub 仓库: 用于存储提示词的公开或私有仓库
GitHub PAT: 具有最低所需权限的 Personal Access Token(详见步骤 2)
Python 3.9+(下面的示例,可以是任何语言),带 FastAPI、Uvicorn、httpx、Pydantic
公共 HTTPS 端点用于你的 webhook 服务器(Render、Fly.io、Heroku 等)

步骤 1:在 Langfuse 中配置提示词 Webhook

在你的 Langfuse 项目中进入 Prompts > Webhooks
点击 Create Webhook
(可选)过滤事件:按接收哪些提示词版本事件的 webhooks 过滤(默认:created、updated、deleted)
设置端点 URL:https:///webhook/prompt
保存并复制 Signing Secret

注意: 你的端点必须返回 2xx 状态码。Langfuse 会以指数退避重试失败的 webhooks。

示例 Webhook 负载

示例 webhook 负载:

json

{
  "id": "550e8400-e29b-41d4-a716-446655440000",
  "timestamp": "2024-07-10T10:30:00Z",
  "type": "prompt-version",
  "action": "created",
  "prompt": {
    "id": "prompt_abc123",
    "name": "movie-critic",
    "version": 3,
    "projectId": "xyz789",
    "labels": ["production", "latest"],
    "prompt": "As a {{criticLevel}} movie critic, rate {{movie}} out of 10.",
    "type": "text",
    "config": { "...": "..." },
    "commitMessage": "Improved critic persona",
    "tags": ["entertainment"],
    "createdAt": "2024-07-10T10:30:00Z",
    "updatedAt": "2024-07-10T10:30:00Z"
  }
}

步骤 2:准备你的 GitHub 仓库和用于同步的 Token

创建一个包含你 GitHub 凭证的 .env 文件:

bash

GITHUB_TOKEN=<your_github_pat_here>
GITHUB_REPO_OWNER=<github_username_or_org>
GITHUB_REPO_NAME=<repo_name>
# (Optional) GITHUB_FILE_PATH=langfuse_prompt.json
# (Optional) GITHUB_BRANCH=main
# (Optional) REQUIRED_LABEL=production

将占位符替换为你的实际值。服务器默认会将提示词提交到 main 分支的 langfuse_prompt.json。如果设置了 REQUIRED_LABEL,只有带该特定标签的提示词才会同步到 GitHub。

用于同步的 GitHub PAT 权限

为了让 webhook 工作,你的 GitHub Personal Access Token 需要最低权限:

权限类型	所需权限
所需权限	Contents:读写,Metadata:只读
旧版 Token Scope	公开仓库:`public_repo` scope,私有仓库:`repo` scope

步骤 3:实现 FastAPI Webhook 服务器

用此 FastAPI 服务器创建 main.py:

python

from typing import Any, Dict
from uuid import UUID
import json
import base64

import httpx
from pydantic import BaseModel, Field
from pydantic_settings import BaseSettings, SettingsConfigDict
from fastapi import FastAPI, HTTPException, Body

class GitHubSettings(BaseSettings):
    """GitHub repository configuration."""
    GITHUB_TOKEN: str
    GITHUB_REPO_OWNER: str
    GITHUB_REPO_NAME: str
    GITHUB_FILE_PATH: str = "langfuse_prompt.json"
    GITHUB_BRANCH: str = "main"
    REQUIRED_LABEL: str = ""  # Optional: only sync prompts with this label

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=True
    )

config = GitHubSettings()

class LangfuseEvent(BaseModel):
    """Langfuse webhook event structure."""
    id: UUID = Field(description="Event identifier")
    timestamp: str = Field(description="Event timestamp")
    type: str = Field(description="Event type")
    action: str = Field(description="Performed action")
    prompt: Dict[str, Any] = Field(description="Prompt content")

async def sync(event: LangfuseEvent) -> Dict[str, Any]:
    """Synchronize prompt data to GitHub repository."""
    # Check if prompt has required label (if specified)
    if config.REQUIRED_LABEL:
        prompt_labels = event.prompt.get("labels", [])
        if config.REQUIRED_LABEL not in prompt_labels:
            return {"skipped": f"Prompt does not have required label '{config.REQUIRED_LABEL}'"}

    api_endpoint = f"https://api.github.com/repos/{config.GITHUB_REPO_OWNER}/{config.GITHUB_REPO_NAME}/contents/{config.GITHUB_FILE_PATH}"

    request_headers = {
        "Authorization": f"Bearer {config.GITHUB_TOKEN}",
        "Accept": "application/vnd.github.v3+json"
    }

    content_json = json.dumps(event.prompt, indent=2)
    encoded_content = base64.b64encode(content_json.encode("utf-8")).decode("utf-8")

    name = event.prompt.get("name", "unnamed")
    version = event.prompt.get("version", "unknown")
    message = f"{event.action}: {name} v{version}"

    payload = {
        "message": message,
        "content": encoded_content,
        "branch": config.GITHUB_BRANCH
    }

    async with httpx.AsyncClient() as http_client:
        try:
            existing = await http_client.get(api_endpoint, headers=request_headers, params={"ref": config.GITHUB_BRANCH})
            if existing.status_code == 200:
                payload["sha"] = existing.json().get("sha")
        except Exception:
            pass

        try:
            response = await http_client.put(api_endpoint, headers=request_headers, json=payload)
            response.raise_for_status()
            return response.json()
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"Repository sync failed: {str(e)}")

app = FastAPI(title="Langfuse GitHub Sync", version="1.0")

@app.post("/webhook/prompt", status_code=201)
async def receive_webhook(event: LangfuseEvent = Body(...)):
    """Process Langfuse webhook and sync to GitHub."""
    result = await sync(event)
    return {
        "status": "synced",
        "commit_info": result.get("commit", {}),
        "file_info": result.get("content", {})
    }

@app.get("/status")
async def health_status():
    """Service health check."""
    return {"healthy": True}

该服务器校验 webhook 负载,在需要时检索现有文件 SHA,并以描述性提交消息将提示词变更提交到 GitHub。

依赖

安装依赖:

bash

pip install fastapi uvicorn pydantic-settings httpx

本地运行

本地运行:

bash

uvicorn main:app --reload --port 8000

在 http://localhost:8000/health 测试健康端点。使用 ngrok 或类似工具暴露 localhost 以进行 webhook 测试。

步骤 4:部署并连接服务器

部署: 使用 Render、Fly.io、Heroku 或类似服务。设置环境变量并确保启用 HTTPS。
更新 Webhook: 在 Langfuse 中,编辑你的 webhook 并将 URL 设置为 https://your-domain.com/webhook/prompt。
测试: 在 Langfuse 中更新一个提示词,并验证你的 GitHub 仓库中出现新的提交。

安全考虑

校验签名: 使用 signing secret 和 x-langfuse-signature header 校验请求
限制 PAT scope: 使用限制到特定仓库的 fine-grained tokens
处理重试: 该实现是幂等的——重复事件不会产生冲突的提交