LangFuse

Weco's LangFuse integration lets you optimize code against LangFuse datasets using both local code evaluators and managed LLM-as-a-Judge evaluators, without writing shell eval scripts.

Instead of writing an evaluation command that prints metrics to stdout, you point Weco at a LangFuse dataset and evaluators. Weco handles the rest: running your target function against each example, collecting scores, and iteratively improving your code.

How it works

1. Target function: A Python function that receives dataset inputs and returns outputs. LangFuse calls this for each item in your dataset.
2. Evaluators: Scoring functions that run locally (code evaluators) and/or LLM-as-a-Judge evaluators configured in the LangFuse UI (managed evaluators).
3. Metric function: A function that combines all evaluator scores into a single number for Weco to optimize.
4. Optimization loop: Weco iteratively modifies your source file, re-runs evaluation against the dataset, and keeps the version that scores best.

Prerequisites

• Python 3.10+
• Weco CLI installed
• A LangFuse account
• API keys for LangFuse and your LLM provider (e.g., OpenAI)

Tutorial: Optimize an HR QA Agent

This walkthrough uses the ZephHR example, a QA agent that answers HR policy questions over fictional documentation. Weco optimizes the agent's prompts to improve answer quality, measured by a composite metric that combines correctness and helpfulness.

git clone https://github.com/WecoAI/weco-cli.git
cd weco-cli/examples/langfuse-zeph-hr-qa

pipx install 'weco[langfuse]'

python -m venv .venv
source .venv/bin/activate
pip install openai langfuse

python -m venv .venv
.venv\Scripts\activate
pip install openai langfuse

python -m venv .venv
.venv\Scripts\Activate.ps1
pip install openai langfuse

export OPENAI_API_KEY="sk-..."
export LANGFUSE_SECRET_KEY="sk-lf-..."
export LANGFUSE_PUBLIC_KEY="pk-lf-..."
export LANGFUSE_BASE_URL="https://cloud.langfuse.com"  # or https://us.cloud.langfuse.com

def answer_hr_question(inputs: dict) -> dict:
    """Answer an HR policy question from the ZephHR docs."""
    question = inputs.get("question", "")

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": USER_TEMPLATE.format(docs=DOCS, question=question)},
        ],
        temperature=0.0,
        response_format={"type": "json_object"},
    )

    # ... parse and return {answer, confidence, relevant_sections}

def qa_score(scores: dict) -> float:
    """Combine correctness (binary gate) with helpfulness (0-1 signal).

    correctness * helpfulness

    - Incorrect answers always score 0.
    - Correct answers are ranked by helpfulness.
    """
    correctness = scores.get("Correctness", 0.0)
    helpfulness = scores.get("Helpfulness", 0.0)
    return correctness * helpfulness

from langfuse import Evaluation

def json_schema_validity(*, input, output, expected_output=None, **kwargs):
    """Check that the agent output contains the required fields."""
    outputs = output or {}
    checks = {
        "answer": isinstance(outputs.get("answer"), str) and len(outputs["answer"]) > 0,
        "confidence": outputs.get("confidence") in ("high", "medium", "low"),
        "relevant_sections": isinstance(outputs.get("relevant_sections"), list),
    }
    passed = all(checks.values())
    failed_fields = [k for k, v in checks.items() if not v]
    comment = "All fields valid" if passed else f"Invalid fields: {', '.join(failed_fields)}"
    return Evaluation(name="json_schema_validity", value=1.0 if passed else 0.0, comment=comment)

def conciseness(*, input, output, expected_output=None, **kwargs):
    """Score based on answer length. Penalises empty or excessively verbose answers."""
    # Returns score: 1.0 for ≤150 words, 0.5 for ≤250, 0.0 for >250 or empty

python setup_dataset.py

weco run --source agent.py \
  --eval-backend langfuse \
  --langfuse-dataset zephhr-qa-opt \
  --langfuse-target agent:answer_hr_question \
  --langfuse-evaluators evaluators:json_schema_validity evaluators:conciseness \
  --langfuse-managed-evaluators Correctness Helpfulness \
  --langfuse-metric-function evaluators:qa_score \
  --additional-instructions optimizer_exemplars.md \
  --metric qa_score --goal maximize --steps 10

weco run --eval-backend langfuse

weco run --source agent.py \
  --eval-backend langfuse \
  --langfuse-dataset zephhr-qa-holdout \
  --langfuse-target agent:answer_hr_question \
  --langfuse-evaluators evaluators:json_schema_validity evaluators:conciseness \
  --langfuse-managed-evaluators Correctness Helpfulness \
  --langfuse-metric-function evaluators:qa_score \
  --metric qa_score --goal maximize --steps 1

weco run --eval-backend langfuse

Key concepts

Target function

Your target function is specified as module:function (e.g., agent:answer_hr_question). It receives an inputs dict from the dataset and returns a dict of outputs. LangFuse calls this function once per dataset item.

Code evaluators

Code evaluators are Python functions specified as module:function (e.g., evaluators:json_schema_validity). Each receives keyword arguments and returns a LangFuse Evaluation object:

from langfuse import Evaluation

def my_evaluator(*, input, output, expected_output=None, **kwargs):
    # input: the dataset item's input dict
    # output: the target function's return value
    # expected_output: the dataset item's expected output (if any)
    return Evaluation(name="my_evaluator", value=1.0, comment="Passed")

Evaluators can also return a plain dict with name/key and value/score keys — the bridge will normalize it to an Evaluation object automatically.

Managed evaluators

Managed evaluators are LLM-as-a-Judge evaluators configured in the LangFuse UI under Evaluation → Evaluators. They run server-side automatically on experiment traces. Weco polls for their scores after each experiment run. Specify them by name with --langfuse-managed-evaluators.

When configuring managed evaluators, make sure the variable mappings point to the correct fields in your traces. For example, if your target function returns {"answer": "...", "confidence": "..."}, map {{generation}} to output.answer in the evaluator template.

Metric function

A metric function combines all evaluator scores into a single number for Weco to optimize. It receives a dict of {evaluator_name: aggregated_score} and returns a float:

def my_metric(scores: dict) -> float:
    return scores["Correctness"] * scores["Helpfulness"]

Summary aggregation

Per-item evaluator scores are aggregated across the dataset using --langfuse-summary (default: mean). Options: mean, median, min, max.

CLI reference

All LangFuse-specific flags for weco run --eval-backend langfuse: