Automating synthetic data labeling for LLM fine-tuning using LangGraph workflows
The biggest issue with basic "LLM-as-a-judge" scripts is the linear flow. If the judge says the response is bad, a standard script just logs it as a fail. With LangGraph, I can build a cyclical correction loop: Generator → Critic → (if fail) → Generator.
Here is the architecture I'm using:
Node A (Generator): Takes a raw seed document and turns it into a Q&A pair.
Node B (Critic): Validates the response against a strict rubric (factuality, brevity, formatting).
Edge (Conditional): If the Critic returns a REJECT signal, the graph loops back to the Generator with the Critic's feedback.
I found that using Claude 3.5 Sonnet for the Critic is non-negotiable. It's significantly better at catching subtle hallucinations than GPT-4o in my tests.
My .cursorrules file is crucial here because I want Cursor to understand the LangGraph state schema whenever I'm adding new nodes. I added this specific instruction:
When modifying LangGraph nodes, always ensure the State typed dictionary is updated first.
Refer to the 'TypedDict' definition in state.py to prevent KeyError during graph execution.Here is a simplified version of the logic I implemented for the conditional edge to handle the "retry" loop:
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
class LabelState(TypedDict):
seed_text: str
synthetic_pair: dict
critique: str
iterations: int
def critic_node(state: LabelState):
# Logic to call LLM and check if synthetic_pair is accurate
# If bad, return {"critique": "too wordy", "iterations": state['iterations'] + 1}
# If good, return {"critique": "PASS"}
pass
def decide_to_retry(state: LabelState) -> Literal["generate", "__end__"]:
if state["critique"] == "PASS" or state["iterations"] > 3:
return "__end__"
return "generate"
workflow = StateGraph(LabelState)
workflow.add_node("generate", generator_node)
workflow.add_node("critic", critic_node)
workflow.set_entry_point("generate")
workflow.add_edge("generate", "critic")
workflow.add_conditional_edges("critic", decide_to_retry)One major gotcha: loop divergence. If your Critic is too picky, you'll hit a maximum iteration limit every time and burn through tokens. I solved this by implementing a "Degradation Prompt"—if the loop hits iteration 2, the Generator is told to prioritize accuracy over style; by iteration 3, it's told to be as concise as possible just to pass the check.
Productivity-wise, this setup allows me to turn a 10-page PDF of technical documentation into 100 high-quality training samples in about 5 minutes. I just pipe the raw text into the graph and export the final state as a JSONL file.
Pro tips for the setup:
Use a separate API key for the Critic so you can track exactly how much "quality control" is costing you versus the actual generation.
Store the critique history in the state. Passing the previous 2 failed attempts back to the Generator prevents it from making the same mistake twice.
Force JSON output using Pydantic objects in the nodes to avoid the classic "Here is the JSON you asked for: ```json ..." wrapper that breaks parsers.
All Replies (0)
No replies yet — be the first!
