Langfuse × GCS プライベートバケットで非公開画像をトレース表示する

with langfuse.start_as_current_observation(
    as_type="generation", name="analyze-image",
    input={"messages": [{"role": "user", "content": [
        {"type": "text", "text": "この画像を分析してください"},
        {"type": "image_url", "image_url": {"url": "https://...画像の URL..."}},
    ]}]},
    model="gemini-2.5-pro",
) as gen:
    result = call_vision_model(...)
    gen.update(output=result)

なぜ自動表示されないのか？
外部の信頼できない URL を自動で読み込むセキュリティリスクを避けるため、意図的にこの挙動になっています（langfuse/langfuse#5030）。
自動レンダリングを設定可能にする Feature Request（#5142）はありますが、現時点では未実装です。

① UI に「Load Image」ボタンが表示される
② クリック → ブラウザが storage.googleapis.com に直接 GET
③ 403 "Anonymous caller does not have storage.objects.get access"

# ❌ これは 403 になる
# url = "https://storage.googleapis.com/your-bucket/path/to/image.png"

# ✅ これなら表示される
url = "https://storage.cloud.google.com/your-bucket/path/to/image.png"

BUCKET_NAME = "your-gcs-bucket"

def gcs_uri_to_authenticated_url(gcs_uri: str) -> str:
    """gs://bucket/key → https://storage.cloud.google.com/bucket/key"""
    path = gcs_uri.removeprefix("gs://")
    return f"https://storage.cloud.google.com/{path}"

authenticated_url = gcs_uri_to_authenticated_url(f"gs://{BUCKET_NAME}/uploads/image.png")

with langfuse.start_as_current_observation(
    as_type="generation", name="analyze-image",
    input={"messages": [{"role": "user", "content": [
        {"type": "text", "text": "この画像を分析してください"},
        {"type": "image_url", "image_url": {"url": authenticated_url}},
    ]}]},
    model="gemini-2.5-pro",
) as gen:
    gen.update(output={"result": "..."})

from langfuse.media import LangfuseMedia

with open("image.png", "rb") as f:
    media = LangfuseMedia(content_bytes=f.read(), content_type="image/png")

with langfuse.start_as_current_observation(
    as_type="generation", name="analyze-image",
    input={"messages": [{"role": "user", "content": [
        {"type": "text", "text": "この画像を分析してください"},
        {"type": "image_url", "image_url": {"url": media}},
    ]}]},
    model="gemini-2.5-pro",
) as gen:
    result = call_vision_model(...)
    gen.update(output=result)

# docker-compose.yml
environment:
  LANGFUSE_S3_MEDIA_UPLOAD_BUCKET: your-gcs-bucket
  LANGFUSE_S3_MEDIA_UPLOAD_REGION: auto
  LANGFUSE_S3_MEDIA_UPLOAD_ACCESS_KEY_ID: <GCS HMAC Access Key>
  LANGFUSE_S3_MEDIA_UPLOAD_SECRET_ACCESS_KEY: <GCS HMAC Secret Key>
  LANGFUSE_S3_MEDIA_UPLOAD_ENDPOINT: https://storage.googleapis.com  # GCS の S3 互換エンドポイント
  LANGFUSE_S3_MEDIA_UPLOAD_FORCE_PATH_STYLE: "true"
  AWS_REQUEST_CHECKSUM_CALCULATION: when_required

`LANGFUSE_S3_MEDIA_UPLOAD_ENDPOINT` について
AWS SDK はデフォルトで AWS S3 に接続します。GCS を使う場合は `https://storage.googleapis.com` を明示的に指定する必要があります。

`AWS_REQUEST_CHECKSUM_CALCULATION` について
Langfuse の AWS SDK v3 はデフォルトで `x-amz-checksum-sha256` ヘッダーを付与しますが、GCS の S3 互換 API はこれを認識せず 400 エラーになります。`when_required` に設定することで回避できます。

ブラウザ → Cloud LB → Cloud Armor（IP制限） → Cloud Run プロキシ → GCS

# main.py
import os
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import Response
from google.cloud import storage

app = FastAPI()
client = storage.Client()
ALLOWED_BUCKETS = set(os.environ.get("ALLOWED_BUCKETS", "").split(","))
ALLOWED_PREFIX = os.environ.get("ALLOWED_PREFIX", "media/")
MAX_UPLOAD_SIZE = int(os.environ.get("MAX_UPLOAD_SIZE", 10 * 1024 * 1024))  # デフォルト 10MB

def _validate(bucket_name: str, object_key: str):
    if bucket_name not in ALLOWED_BUCKETS:
        raise HTTPException(status_code=403, detail="Bucket not allowed")
    if not object_key.startswith(ALLOWED_PREFIX):
        raise HTTPException(status_code=403, detail="Path not allowed")

@app.get("/{bucket_name}/{object_key:path}")
def get_object(bucket_name: str, object_key: str):
    """ブラウザからの画像取得リクエストを GCS に転送する"""
    _validate(bucket_name, object_key)
    bucket = client.bucket(bucket_name)
    blob = bucket.blob(object_key)
    if not blob.exists():
        raise HTTPException(status_code=404, detail="Not found")
    content = blob.download_as_bytes()
    return Response(content=content, media_type=blob.content_type or "application/octet-stream")

@app.put("/{bucket_name}/{object_key:path}")
async def put_object(bucket_name: str, object_key: str, request: Request):
    """LangfuseMedia からのアップロードリクエストを GCS に転送する"""
    _validate(bucket_name, object_key)
    content_length = int(request.headers.get("content-length", 0))
    if content_length > MAX_UPLOAD_SIZE:
        raise HTTPException(status_code=413, detail="File too large")
    content = await request.body()
    if len(content) > MAX_UPLOAD_SIZE:
        raise HTTPException(status_code=413, detail="File too large")
    content_type = request.headers.get("content-type", "application/octet-stream")
    bucket = client.bucket(bucket_name)
    blob = bucket.blob(object_key)
    blob.upload_from_string(content, content_type=content_type)
    return Response(status_code=200)

# requirements.txt
fastapi
uvicorn
google-cloud-storage

補足：`--source` デプロイでは Buildpacks が `main.py` と `requirements.txt` を自動検出するため、Dockerfile は不要です。

```bash
export PROJECT_ID=your-project
export BUCKET_NAME=your-gcs-bucket

# サービスアカウント作成・権限付与
gcloud iam service-accounts create langfuse-media-proxy-sa \
  --project="${PROJECT_ID}"

gcloud storage buckets add-iam-policy-binding "gs://${BUCKET_NAME}" \
  --member="serviceAccount:langfuse-media-proxy-sa@${PROJECT_ID}.iam.gserviceaccount.com" \
  --role="roles/storage.objectAdmin"

# デプロイ（ingress 制限 + デフォルト URL 無効化を同時に設定）
gcloud run deploy langfuse-media-proxy \
  --source=. \
  --region=asia-northeast1 \
  --service-account="langfuse-media-proxy-sa@${PROJECT_ID}.iam.gserviceaccount.com" \
  --allow-unauthenticated \
  --ingress=internal-and-cloud-load-balancing \
  --no-default-url \
  --set-env-vars="ALLOWED_BUCKETS=${BUCKET_NAME}" \
  --port=8080

# Serverless NEG
gcloud compute network-endpoint-groups create langfuse-proxy-neg \
  --region=asia-northeast1 \
  --network-endpoint-type=serverless \
  --cloud-run-service=langfuse-media-proxy \
  --project="${PROJECT_ID}"

# バックエンドサービス
gcloud compute backend-services create langfuse-proxy-backend \
  --global \
  --project="${PROJECT_ID}"

gcloud compute backend-services add-backend langfuse-proxy-backend \
  --global \
  --network-endpoint-group=langfuse-proxy-neg \
  --network-endpoint-group-region=asia-northeast1 \
  --project="${PROJECT_ID}"

# URL マップ・HTTPS プロキシ・転送ルール
gcloud compute url-maps create langfuse-proxy-lb \
  --default-service=langfuse-proxy-backend \
  --project="${PROJECT_ID}"

gcloud compute target-https-proxies create langfuse-proxy-https \
  --url-map=langfuse-proxy-lb \
  --ssl-certificates=your-ssl-cert \
  --project="${PROJECT_ID}"

gcloud compute forwarding-rules create langfuse-proxy-rule \
  --target-https-proxy=langfuse-proxy-https \
  --ports=443 \
  --global \
  --project="${PROJECT_ID}"

# Cloud Armor セキュリティポリシー
gcloud compute security-policies create langfuse-media-policy \
  --project="${PROJECT_ID}"

gcloud compute security-policies rules update 2147483647 \
  --security-policy=langfuse-media-policy \
  --action=deny-403 \
  --project="${PROJECT_ID}"

gcloud compute security-policies rules create 1000 \
  --security-policy=langfuse-media-policy \
  --expression="inIpRange(origin.ip, 'YOUR_IP/32')" \
  --action=allow \
  --project="${PROJECT_ID}"

gcloud compute backend-services update langfuse-proxy-backend \
  --security-policy=langfuse-media-policy \
  --global \
  --project="${PROJECT_ID}"

gcloud storage buckets create "gs://${BUCKET_NAME}" \
  --uniform-bucket-level-access \
  --pap \
  --project="${PROJECT_ID}"

gcloud storage buckets add-iam-policy-binding "gs://${BUCKET_NAME}" \
  --member="serviceAccount:langfuse-media-proxy-sa@${PROJECT_ID}.iam.gserviceaccount.com" \
  --role="roles/storage.objectAdmin"

PROXY_BASE_URL = "https://your-proxy-domain.com"
BUCKET_NAME = "your-gcs-bucket"

def gcs_uri_to_proxy_url(gcs_uri: str) -> str:
    path = gcs_uri.removeprefix("gs://")
    return f"{PROXY_BASE_URL}/{path}"

```yaml
# docker-compose.yml
environment:
  LANGFUSE_S3_MEDIA_UPLOAD_BUCKET: your-gcs-bucket
  LANGFUSE_S3_MEDIA_UPLOAD_REGION: auto
  LANGFUSE_S3_MEDIA_UPLOAD_ACCESS_KEY_ID: <GCS HMAC Access Key>
  LANGFUSE_S3_MEDIA_UPLOAD_SECRET_ACCESS_KEY: <GCS HMAC Secret Key>
  LANGFUSE_S3_MEDIA_UPLOAD_ENDPOINT: https://your-proxy-domain.com
  LANGFUSE_S3_MEDIA_UPLOAD_FORCE_PATH_STYLE: "true"
  LANGFUSE_S3_MEDIA_UPLOAD_PREFIX: media/
  AWS_REQUEST_CHECKSUM_CALCULATION: when_required