Skip to content

UCCA Agent Workflow

Jump to bottom
TenzinGayche edited this page Dec 17, 2025 · 1 revision

UCCA Agent Workflow

The UCCA Agent Workflow (ucca_agent.py) implements an iterative generation-evaluation loop using LangGraph. This is the advanced workflow that uses scholarly commentaries to refine UCCA graphs.

🏗️ Architecture

┌─────────────────────────────────────────────────────────────────────────┐
│                         LangGraph StateGraph                             │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                          │
│   ┌─────────┐                                                           │
│   │  START  │                                                           │
│   └────┬────┘                                                           │
│        │                                                                │
│        ▼                                                                │
│   ┌────────────────────────────────────────────────────────────┐       │
│   │                    ucca_generator                           │       │
│   │                                                             │       │
│   │  • First call: Uses get_ucca_generation_prompt()           │       │
│   │  • Subsequent: Uses feedback-based refinement prompt       │       │
│   │  • Adds new graph to source_ucca list                      │       │
│   │  • Increments iteration counter                            │       │
│   └────────────────────────────┬───────────────────────────────┘       │
│                                │                                        │
│                                ▼                                        │
│   ┌────────────────────────────────────────────────────────────┐       │
│   │                    ucca_evaluator                           │       │
│   │                                                             │       │
│   │  • Uses get_source_ucca_refinement_prompt()                │       │
│   │  • Evaluates against commentaries                          │       │
│   │  • Returns StructuredFeedback with grade                   │       │
│   │  • Adds suggestions to latest_structured_suggestions       │       │
│   └────────────────────────────┬───────────────────────────────┘       │
│                                │                                        │
│                                ▼                                        │
│                        ┌──────────────┐                                │
│                        │  route_ucca  │                                │
│                        └──────┬───────┘                                │
│                               │                                         │
│              ┌────────────────┼────────────────┐                       │
│              │                │                │                        │
│              ▼                ▼                │                        │
│     grade == "great"    itteration >= 1       │                        │
│              │                │                │                        │
│              └────────┬───────┘                │                        │
│                       │                        │                        │
│                       ▼                        ▼                        │
│                  "Accepted"            "Rejected + Feedback"           │
│                       │                        │                        │
│                       ▼                        │                        │
│                  ┌─────────┐                   │                        │
│                  │   END   │                   │                        │
│                  └─────────┘                   │                        │
│                                                │                        │
│   ┌────────────────────────────────────────────┘                       │
│   │   (loops back to ucca_generator with feedback)                     │
│   └─────────────────────────────────────────────                       │
│                                                                          │
└─────────────────────────────────────────────────────────────────────────┘

🔧 Workflow Construction

The workflow is built using LangGraph's StateGraph:

from langgraph.graph import StateGraph, START, END

optimizer_builder = StateGraph(State)

# Add nodes
optimizer_builder.add_node("ucca_generator", ucca_generator)
optimizer_builder.add_node("ucca_evaluator", ucca_evaluator)

# Add edges
optimizer_builder.add_edge(START, "ucca_generator")
optimizer_builder.add_edge("ucca_generator", "ucca_evaluator")

# Add conditional routing
optimizer_builder.add_conditional_edges(
    "ucca_evaluator",
    route_ucca,
    {
        "Accepted": END,
        "Rejected + Feedback": "ucca_generator"
    }
)

# Compile the workflow
optimizer_workflow = optimizer_builder.compile()

📦 Node Functions

1. ucca_generator(state: State)

Generates or refines UCCA graphs based on current state.

First Iteration (no feedback):

  • Uses get_ucca_generation_prompt()
  • Creates initial UCCA graph from source text
  • Returns {"source_ucca": [graph], "itteration": 0}

Subsequent Iterations (with feedback):

  • Uses refinement prompt with latest feedback
  • Addresses issues from evaluation
  • Returns {"source_ucca": state["source_ucca"] + [new_graph], "itteration": current + 1} 
def ucca_generator(state: State):
    if state.get("feedback_history"):
        # Refinement mode - use feedback
        latest_feedback = state["feedback_history"][-1]
        prompt = f"""
        SYSTEM: You are a UCCA graph refinement expert...
        
        Source Text: {state['source']}
        Feedback to Address: {latest_feedback}
        Current UCCA Graph: {ucca_graph_to_text(state['source_ucca'][-1])}
        
        # REQUIREMENTS
        - Address EVERY issue in feedback
        - Generate COMPLETE, VALID JSON
        ...
        """
        # Generate refined graph
    else:
        # Initial generation
        prompt = get_ucca_generation_prompt(state["source"], "Tibetan")
        # Generate initial graph

2. ucca_evaluator(state: State)

Evaluates the latest UCCA graph against scholarly commentaries.

Process:

  1. Gets latest graph from state["source_ucca"][-1]
  2. Builds evaluation prompt with commentaries
  3. Invokes LLM with StructuredFeedback output schema
  4. Extracts grade and suggestions
  5. Updates feedback history
def ucca_evaluator(state: State):
    # Skip if already at max iterations
    if state["itteration"] >= 1:
        return {"itteration": state["itteration"]}
    
    latest_ucca_graph = state["source_ucca"][-1]
    
    prompt_text = get_source_ucca_refinement_prompt(
        source_text=state["source"],
        current_ucca_analysis=latest_ucca_graph,
        commentaries=[state["commentary1"], state["commentary2"], state["commentary3"]],
        sanskrit=state["sanskrit"],
        language=state["language"]
    )
    
    # Get structured evaluation
    result = llm.with_structured_output(StructuredFeedback).invoke(prompt_text)
    
    return {
        "grade": result.grade,
        "feedback_history": [...updated...],
        "latest_structured_suggestions": result.specific_suggestions,
        "itteration": current_iteration
    }

3. route_ucca(state: State)

Determines whether to accept the graph or continue refining.

def route_ucca(state: State):
    if state["grade"] == "great":
        return "Accepted"
    elif state["itteration"] >= 1:  # Max iterations reached
        return "Accepted"
    else:
        return "Rejected + Feedback"

Routing Logic:

Condition Route Outcome
grade == "great" Accepted End workflow
itteration >= 1 Accepted End (max iterations)
Otherwise Rejected + Feedback Loop back for refinement

🛠️ Utility Functions

ucca_graph_to_text(ucca_graph: UCCAGraph) -> str

Converts a UCCA graph to human-readable text format.

def ucca_graph_to_text(ucca_graph: UCCAGraph) -> str:
    """
    Output example:
    UCCA Graph (Root ID: 0)
    |- [0] Type: Scene - "སྔོན་ཆད་མ་བྱུང་བ་ཡང་འདིར་བརྗོད་མེད། །"
      |- [1] Type: Implicit Participant-Agent - ""
      |- [2] Type: Process - "བརྗོད་མེད"
      |- [3] Type: Participant-Location - "འདིར"
    Total nodes: 4
    """

display_ucca_graph(state: State)

Prints the latest UCCA graph from state.

def display_ucca_graph(state: State):
    if state.get("source_ucca") and len(state["source_ucca"]) > 0:
        latest_graph = state["source_ucca"][-1]
        print(ucca_graph_to_text(latest_graph))

📊 State Flow

Initial State
    │
    ├── source: "Tibetan text..."
    ├── commentary1/2/3: "Commentary texts..."
    ├── source_ucca: []
    ├── feedback_history: []
    ├── itteration: 0
    └── grade: ""
          │
          ▼
    [ucca_generator - Initial]
          │
    ├── source_ucca: [Graph1]
    └── itteration: 0
          │
          ▼
    [ucca_evaluator]
          │
    ├── grade: "okay"
    ├── feedback_history: ["Iteration 0 - Grade: okay\n..."]
    └── latest_structured_suggestions: [...]
          │
          ▼
    [route_ucca] → "Rejected + Feedback"
          │
          ▼
    [ucca_generator - Refinement]
          │
    ├── source_ucca: [Graph1, Graph2]
    └── itteration: 1
          │
          ▼
    [ucca_evaluator]
          │
          ▼
    [route_ucca] → "Accepted" (itteration >= 1)
          │
          ▼
    END

🚀 Usage

from ucca_generator.ucca_agent import optimizer_workflow

# Prepare initial state
initial_state = {
    "source": "མི་རྟག་པ་སྒོམ་པ་ལ་དགོས་པ་ཅི་ཡོད།",
    "sanskrit": "अनित्यता भावना",
    "language": "Tibetan",
    "feedback_history": [],
    "commentary1": "This verse discusses the purpose of meditation...",
    "commentary2": "The practitioner should understand...",
    "commentary3": "According to the tradition...",
    "itteration": 0,
    "format_iteration": 0,
    "source_ucca": [],
    "latest_structured_suggestions": [],
    "grade": ""
}

# Run the workflow
result = optimizer_workflow.invoke(initial_state)

# Access results
final_graph = result["source_ucca"][-1]
final_grade = result["grade"]
all_feedback = result["feedback_history"]

print(f"Final grade: {final_grade}")
print(f"Total iterations: {len(result['source_ucca'])}")

⚙️ Configuration

Max Iterations

Currently hardcoded to 1 in route_ucca():

elif state["itteration"] >= 1:  # Change this value
    return "Accepted"

LLM Model

llm = ChatAnthropic(model="claude-3-7-sonnet-latest", max_tokens=8000)

🔍 Debugging

View Graph at Any Point

from ucca_generator.ucca_agent import display_ucca_graph

# During workflow or after
display_ucca_graph(result)

Check Feedback History

for i, feedback in enumerate(result["feedback_history"]):
    print(f"=== Iteration {i} ===")
    print(feedback)
    print()

⚠️ Important Notes

  1. Commentaries are required for meaningful evaluation
  2. Iteration limit is currently set to 1 (one refinement cycle)
  3. Error handling includes retry logic for LLM failures
  4. Graph history is preserved in source_ucca list for comparison
  5. Structured output ensures consistent feedback format

Clone this wiki locally