Feedback Loops

This guide covers the complete feedback loop: capturing expert signals on pipeline outputs, running synthesis to generate improvements, and applying those improvements.

Overview

The feedback loop workflow consists of four steps:

1. Capture signals - Record expert feedback on pipeline outputs
2. Accumulate feedback - Gather enough signals to identify patterns
3. Run synthesis - Generate AI-proposed improvements
4. Apply and verify - Implement changes and measure impact

Click to expand

Step 1: Capture Signals

When an expert reviews a pipeline execution, capture their feedback as a signal:

Rating Signal

For quick assessments:

Capture rating signal

cURL

curl -X POST https://api.catalyzed.ai/signals \
  -H "Authorization: Bearer $API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "teamId": "ZkoDMyjZZsXo4VAO_nJLk",
    "data": {
      "type": "rating",
      "score": 0.6
    },
    "executionIds": ["GkR8I6rHBms3W4Qfa2-FN"]
  }'

TypeScript

await fetch("https://api.catalyzed.ai/signals", {
  method: "POST",
  headers: {
    Authorization: `Bearer ${apiToken}`,
    "Content-Type": "application/json",
  },
  body: JSON.stringify({
    teamId: "ZkoDMyjZZsXo4VAO_nJLk",
    data: { type: "rating", score: 0.6 },
    executionIds: ["GkR8I6rHBms3W4Qfa2-FN"],
  }),
});

Python

requests.post(
    "https://api.catalyzed.ai/signals",
    headers={"Authorization": f"Bearer {api_token}"},
    json={
        "teamId": "ZkoDMyjZZsXo4VAO_nJLk",
        "data": {"type": "rating", "score": 0.6},
        "executionIds": ["GkR8I6rHBms3W4Qfa2-FN"]
    }
)

Comment Signal

For detailed feedback:

Capture comment signal

cURL

curl -X POST https://api.catalyzed.ai/signals \
  -H "Authorization: Bearer $API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "teamId": "ZkoDMyjZZsXo4VAO_nJLk",
    "data": {
      "type": "comment",
      "text": "The summary is missing the revenue growth percentage. Should always include key financial metrics."
    },
    "executionIds": ["GkR8I6rHBms3W4Qfa2-FN"]
  }'

TypeScript

await fetch("https://api.catalyzed.ai/signals", {
  method: "POST",
  headers: {
    Authorization: `Bearer ${apiToken}`,
    "Content-Type": "application/json",
  },
  body: JSON.stringify({
    teamId: "ZkoDMyjZZsXo4VAO_nJLk",
    data: {
      type: "comment",
      text: "The summary is missing the revenue growth percentage. Should always include key financial metrics.",
    },
    executionIds: ["GkR8I6rHBms3W4Qfa2-FN"],
  }),
});

Python

requests.post(
    "https://api.catalyzed.ai/signals",
    headers={"Authorization": f"Bearer {api_token}"},
    json={
        "teamId": "ZkoDMyjZZsXo4VAO_nJLk",
        "data": {
            "type": "comment",
            "text": "The summary is missing the revenue growth percentage. Should always include key financial metrics."
        },
        "executionIds": ["GkR8I6rHBms3W4Qfa2-FN"]
    }
)

Writing Effective Comments

Good comments are:

• Specific - Point to exactly what was wrong
• Actionable - Explain what the correct behavior would be
• Consistent - Use similar language for similar issues

Step 2: Accumulate Feedback

Gather enough signals to identify patterns. The synthesis process works best with:

• 5-10 signals minimum for initial patterns
• 20+ signals for robust improvements
• Diverse examples covering different failure modes

Check Signal Coverage

Review accumulated signals for a pipeline:

List signals for pipeline

cURL

# Get executions for pipeline
EXECUTIONS=$(curl -s "https://api.catalyzed.ai/pipeline-executions?pipelineId=$PIPELINE_ID" \
  -H "Authorization: Bearer $API_TOKEN" | jq -r '.executions[].executionId' | paste -sd,)

# Get signals for those executions
curl "https://api.catalyzed.ai/signals?executionIds=$EXECUTIONS" \
  -H "Authorization: Bearer $API_TOKEN"

TypeScript

// Get recent executions
const execResponse = await fetch(
  `https://api.catalyzed.ai/pipeline-executions?pipelineId=${pipelineId}&pageSize=50`,
  { headers: { Authorization: `Bearer ${apiToken}` } }
);
const { executions } = await execResponse.json();
const executionIds = executions.map((e: any) => e.executionId).join(",");

// Get signals for those executions
const signalsResponse = await fetch(
  `https://api.catalyzed.ai/signals?executionIds=${executionIds}`,
  { headers: { Authorization: `Bearer ${apiToken}` } }
);
const { signals } = await signalsResponse.json();

console.log(`Total signals: ${signals.length}`);
console.log(`Comments: ${signals.filter((s: any) => s.dataType === "comment").length}`);
console.log(`Ratings: ${signals.filter((s: any) => s.dataType === "rating").length}`);

Python

# Get recent executions
exec_response = requests.get(
    "https://api.catalyzed.ai/pipeline-executions",
    params={"pipelineId": pipeline_id, "pageSize": 50},
    headers={"Authorization": f"Bearer {api_token}"}
)
executions = exec_response.json()["executions"]
execution_ids = ",".join(e["executionId"] for e in executions)

# Get signals for those executions
signals_response = requests.get(
    "https://api.catalyzed.ai/signals",
    params={"executionIds": execution_ids},
    headers={"Authorization": f"Bearer {api_token}"}
)
signals = signals_response.json()["signals"]

print(f"Total signals: {len(signals)}")
print(f"Comments: {len([s for s in signals if s['dataType'] == 'comment'])}")
print(f"Ratings: {len([s for s in signals if s['dataType'] == 'rating'])}")

Step 3: Run Synthesis

When you have enough signals, run synthesis to generate improvement proposals:

Run synthesis

cURL

curl -X POST "https://api.catalyzed.ai/pipelines/$PIPELINE_ID/synthesize" \
  -H "Authorization: Bearer $API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "handlerType": "apg_v1"
  }'

TypeScript

const synthResponse = await fetch(
  `https://api.catalyzed.ai/pipelines/${pipelineId}/synthesize`,
  {
    method: "POST",
    headers: {
      Authorization: `Bearer ${apiToken}`,
      "Content-Type": "application/json",
    },
    body: JSON.stringify({ handlerType: "apg_v1" }),
  }
);
const synthesisRun = await synthResponse.json();
console.log(`Synthesis started: ${synthesisRun.synthesisRunId}`);

Python

synth_response = requests.post(
    f"https://api.catalyzed.ai/pipelines/{pipeline_id}/synthesize",
    headers={"Authorization": f"Bearer {api_token}"},
    json={"handlerType": "apg_v1"}
)
synthesis_run = synth_response.json()
print(f"Synthesis started: {synthesis_run['synthesisRunId']}")

Expected Duration: Synthesis typically completes in 30-90 seconds, depending on:

• Number of signals being analyzed (10-20 signals: ~30s, 50+ signals: ~90s)
• Complexity of the pipeline configuration
• Current system load

Wait for Completion

async function waitForSynthesis(synthesisRunId: string) {
  while (true) {
    const response = await fetch(
      `https://api.catalyzed.ai/synthesis-runs/${synthesisRunId}`,
      { headers: { Authorization: `Bearer ${apiToken}` } }
    );
    const run = await response.json();

    if (run.status === "generated") {
      return run;
    }
    if (run.status === "failed") {
      throw new Error(run.errorMessage);
    }

    console.log(`Status: ${run.status}...`);
    await new Promise(r => setTimeout(r, 3000));
  }
}

const completedSynth = await waitForSynthesis(synthesisRun.synthesisRunId);

Review the Proposal

The completed synthesis includes a diagnosis and proposed changes:

{
  "synthesisRunId": "SynR8I6rHBms3W4Qfa2-FN",
  "status": "generated",
  "diagnosis": {
    "summary": "Users report summaries are missing key financial metrics and lack consistent formatting.",
    "findings": [
      {
        "id": "1",
        "type": "pain_point",
        "description": "Summaries frequently omit revenue growth percentages",
        "signalIds": ["SigXYZ1...", "SigXYZ2..."],
        "severity": "high"
      }
    ]
  },
  "changes": [
    {
      "id": "1",
      "type": "modify",
      "section": "configuration",
      "slotId": "system_prompt",
      "description": "Updated system prompt to emphasize financial metrics",
      "rationale": "Addresses finding #1"
    }
  ],
  "structuredWarnings": [
    {
      "type": "assumption",
      "message": "Assumed all documents contain financial metrics"
    }
  ],
  "confidence": "high"
}

Step 4: Apply and Verify

Review Warnings

Before applying, check the structuredWarnings:

Warning Type	Meaning	Action
breaking_change	May affect existing behavior	Test thoroughly
assumption	Synthesis made an assumption	Verify it’s correct
tradeoff	Change has pros and cons	Consider impact
needs_review	Requires human judgment	Review carefully

Apply the Changes

Apply synthesis

cURL

curl -X POST "https://api.catalyzed.ai/synthesis-runs/$SYNTHESIS_RUN_ID/apply" \
  -H "Authorization: Bearer $API_TOKEN"

TypeScript

await fetch(`https://api.catalyzed.ai/synthesis-runs/${synthesisRunId}/apply`, {
  method: "POST",
  headers: { Authorization: `Bearer ${apiToken}` },
});
console.log("Changes applied!");

Python

requests.post(
    f"https://api.catalyzed.ai/synthesis-runs/{synthesis_run_id}/apply",
    headers={"Authorization": f"Bearer {api_token}"}
)
print("Changes applied!")

Or Reject If Not Appropriate

curl -X POST "https://api.catalyzed.ai/synthesis-runs/$SYNTHESIS_RUN_ID/reject" \
  -H "Authorization: Bearer $API_TOKEN"

Verify Improvements

After applying, run an evaluation to measure improvement:

// Run evaluation against existing example set
const evaluation = await runEvaluation(pipelineId, exampleSetId);
const completed = await waitForEvaluation(evaluation.evaluationId);

console.log(`New score: ${completed.aggregateScore}`);
console.log(`Passed: ${completed.passedCount}/${completed.totalExamples}`);

Complete Workflow Example

Here’s the full feedback loop in one script:

async function runFeedbackLoop(pipelineId: string) {
  // 1. Check if we have enough signals
  const signals = await getSignalsForPipeline(pipelineId);
  if (signals.length < 5) {
    console.log(`Only ${signals.length} signals. Collect more feedback first.`);
    return;
  }

  // 2. Run synthesis
  console.log(`Running synthesis with ${signals.length} signals...`);
  const synthResponse = await fetch(
    `https://api.catalyzed.ai/pipelines/${pipelineId}/synthesize`,
    {
      method: "POST",
      headers: { Authorization: `Bearer ${apiToken}`, "Content-Type": "application/json" },
      body: JSON.stringify({ handlerType: "apg_v1" }),
    }
  );
  const synthRun = await synthResponse.json();

  // 3. Wait for completion
  const completed = await waitForSynthesis(synthRun.synthesisRunId);

  // 4. Show proposal summary
  console.log("\n=== Synthesis Results ===");
  console.log(`Diagnosis: ${completed.diagnosis.summary}`);
  console.log(`Findings: ${completed.diagnosis.findings.length}`);
  console.log(`Proposed changes: ${completed.changes.length}`);
  console.log(`Confidence: ${completed.confidence}`);

  if (completed.structuredWarnings.length > 0) {
    console.log("\nWarnings:");
    for (const w of completed.structuredWarnings) {
      console.log(`  - [${w.type}] ${w.message}`);
    }
  }

  // 5. Apply if high confidence (or prompt for human review)
  if (completed.confidence === "high") {
    console.log("\nApplying changes...");
    await fetch(`https://api.catalyzed.ai/synthesis-runs/${completed.synthesisRunId}/apply`, {
      method: "POST",
      headers: { Authorization: `Bearer ${apiToken}` },
    });
    console.log("Done! Run an evaluation to verify improvements.");
  } else {
    console.log("\nReview the proposal and apply manually if appropriate.");
  }
}

Best Practices

1. Capture feedback consistently - Establish team conventions for ratings and comments
2. Be specific in comments - Vague feedback leads to vague improvements
3. Review before applying - Always check warnings and understand proposed changes
4. Verify with evaluations - Measure before and after to confirm improvement
5. Iterate - Run multiple cycles to continuously improve

Related Guides

• Signals - Signal types and API reference
• Synthesis Runs - Synthesis process details
• Evaluation Workflow - Measuring pipeline quality
• Promoting Executions - Build example sets from feedback