Conditional output bug

[endymion]

LangGraphScore Conditional Outputs Override by Top-Level Output Mappings

Summary

LangGraphScore configurations with conditional outputs are being overridden by top-level output mappings, causing incorrect score results. When a node's conditions should trigger specific output values (e.g., value: "Yes"), the final result shows the classification value instead (e.g., value: "Computer_With_Internet").

Environment

• Plexus Version: Latest main branch
• Python Version: 3.11
• Affected Component: plexus/scores/LangGraphScore.py

Bug Description

Expected Behavior

When a LangGraphScore has a node with conditions that route to END and specify conditional outputs, those outputs should be preserved in the final result, taking precedence over any top-level output mappings.

Actual Behavior

Top-level output mappings (e.g., value: classification) override conditional outputs, causing incorrect results to be returned.

Reproduction Case

Score Configuration

# Simplified example based on "Computer w/ Internet" score (scorecard 1039, score 45417)
output:
  value: classification  # This is overriding conditional outputs

nodes:
  device_response_analyzer:
    type: Classifier
    prompt: "Classify if device has internet access"
    classes:
      - Computer_With_Internet
      - Computer_Without_Internet
      - Mobile_Device
    conditions:
      Computer_With_Internet:
        target: END
        output:
          value: "Yes"  # This should be the final value
          explanation: "Computer with internet access detected"

Test Input

Any input that gets classified as "Computer_With_Internet"

Expected Output

{
  "value": "Yes",
  "explanation": "Computer with internet access detected",
  "classification": "Computer_With_Internet"
}

Actual Output

{
  "value": "Computer_With_Internet",  // ❌ Wrong - should be "Yes"
  "explanation": "Computer with internet access detected",
  "classification": "Computer_With_Internet"
}

Steps to Reproduce

1. Create a LangGraphScore with:

   • A node that has conditions routing to END
   • Conditional outputs that set specific values
   • Top-level output mapping (e.g., value: classification)

2. Run a prediction that triggers the condition

3. Observe that the final result uses the top-level output mapping instead of the conditional output

Technical Analysis

Root Cause

In plexus/scores/LangGraphScore.py, the add_edges method appears to have issues with conditional routing logic:

1. Final Node Conditions: Conditions on final nodes may not be properly processed
2. Conditional Routing: The routing from conditional nodes to END may not preserve conditional outputs
3. Output Priority: Top-level output mappings are taking precedence over conditional outputs

Code Location

The issue is in the add_edges method of the LangGraphScore class, specifically around line areas that handle:

• Conditional edge creation
• Final node processing
• Output aliasing logic

Impact

Severity: High

This bug affects the core functionality of LangGraphScore conditional logic, causing incorrect classification results.

Affected Scores

Any LangGraphScore that uses:

• Conditional routing to END
• Conditional outputs that differ from top-level mappings
• Multiple nodes with conditions

Business Impact

• Incorrect score results in production
• Potential misclassification of important data
• Unreliable conditional logic in complex scoring workflows

Additional Context

Real-World Example

The "Computer w/ Internet" score in production is affected:

• Scorecard ID: 1039
• Score ID: 45417
• Symptom: Returns "Computer_With_Internet" instead of "Yes"

Testing Suggestions

1. Create unit tests with single-node conditional scenarios
2. Test multi-node workflows with conditions on middle nodes
3. Test final node conditions routing to END
4. Verify output precedence rules

Debugging Information

When testing predictions with the affected score:

# Using MCP service to test
plexus_predict(
    scorecard_name="Computer w/ Internet", 
    score_name="Computer w/ Internet",
    item_id="test_item"
)

Proposed Investigation Steps

1. Code Review: Examine the add_edges method for conditional routing logic
2. Debug Logging: Add comprehensive logging to trace output value precedence
3. Unit Tests: Create regression tests for various conditional scenarios
4. Integration Testing: Test with real score configurations

Related Files

• plexus/scores/LangGraphScore.py - Main implementation
• plexus/scores/LangGraphScore_test.py - Test file for adding regression tests

Labels Suggestion

• bug
• high-priority
• scoring
• langgraph

---

Note: This bug affects the reliability of conditional scoring logic and should be prioritized for investigation and resolution.

[endymion]

Update: Now we have a unit test for reproducing this in https://github.com/AnthusAI/Plexus/tree/bugfix/conditional-output-bug-80

FYI @dereknorrbom

Plexus/plexus/scores/LangGraphScore_test.py

Lines 9849 to 10002 in 763fb54

	async def test_production_conditional_output_override_bug():
	"""
	Test that reproduces the exact production bug where conditional outputs
	are overridden by top-level output mappings.
	This test matches the exact production configuration:
	- No node-level output mapping (commented out in production)
	- Real Classifier behavior simulation
	- Conditional routing to END with output value: "Yes"
	- Top-level output mapping value: classification (the bug!)
	Expected: value should be "Yes" from conditional output
	Actual: value becomes "Computer_With_Internet" from top-level mapping
	"""
	# Mock a Classifier that behaves exactly like the real one
	class ProductionClassifierMock:
	def __init__(self, **kwargs):
	self.parameters = MagicMock()
	# Critical: NO node-level output mapping (matches production)
	self.parameters.output = None
	self.parameters.name = kwargs.get('name', 'device_response_analyzer')
	class GraphState(BaseModel):
	model_config = ConfigDict(arbitrary_types_allowed=True)
	text: Optional[str] = None
	metadata: Optional[dict] = None
	results: Optional[dict] = None
	classification: Optional[str] = None
	explanation: Optional[str] = None
	# Fields that conditional outputs will set
	value: Optional[str] = None
	device_status: Optional[str] = None
	prospect_response: Optional[str] = None
	async def analyze(self, state):
	# Simulate exact production classification behavior
	state_dict = state.model_dump() if hasattr(state, 'model_dump') else dict(state)
	state_dict.update({
	"classification": "Computer_With_Internet",
	"explanation": "Prospect Response: Yes. Answer: Computer_With_Internet"
	})
	return self.GraphState(**state_dict)
	def build_compiled_workflow(self, graph_state_class):
	from langgraph.graph import StateGraph, END
	workflow = StateGraph(graph_state_class)
	workflow.add_node("analyze", self.analyze)
	workflow.set_entry_point("analyze")
	workflow.add_edge("analyze", END)
	async def invoke_workflow(state):
	if hasattr(state, 'model_dump'):
	state_dict = state.model_dump()
	else:
	state_dict = dict(state)
	final_state = await self.analyze(graph_state_class(**state_dict))
	return final_state.model_dump()
	return invoke_workflow
	# Mock the import
	original_import = LangGraphScore._import_class
	@staticmethod
	def mock_import_class(class_name):
	if class_name == "ProductionClassifierMock":
	return ProductionClassifierMock
	else:
	return original_import(class_name)
	try:
	LangGraphScore._import_class = mock_import_class
	# Configuration that EXACTLY matches production structure
	config = {
	'name': 'Production Conditional Override Bug Reproduction',
	'model_provider': 'ChatOpenAI',
	'model_name': 'gpt-4o-mini-2024-07-18',
	'graph': [
	{
	'name': 'device_response_analyzer',
	'class': 'ProductionClassifierMock',
	'valid_classes': [
	'Computer_With_Internet',
	'Computer_No_Internet',
	'Internet_No_Computer',
	'Invalid_Device',
	'No_Device',
	'Library',
	'NoResponse'
	],
	# Critical: NO node-level output mapping (matches production)
	# In production this is commented out:
	# output:
	# device_status: classification
	# prospect_response: explanation
	'conditions': [
	{
	'state': 'classification',
	'value': 'Computer_With_Internet',
	'node': 'END',
	'output': {
	'value': 'Yes', # This should be preserved
	'explanation': 'The prospect said they have a computer with internet, this is a yes.',
	'device_status': 'classification',
	'prospect_response': 'explanation'
	}
	}
	]
	}
	],
	# The problematic top-level output mapping that overrides conditionals
	'output': {
	'value': 'classification', # THIS IS THE BUG!
	'explanation': 'explanation'
	}
	}
	# Create the score
	score = await LangGraphScore.create(**config)
	# Create test input that triggers the condition
	test_input = Score.Input(
	text="Agent: Do you have access to a computer with Internet access? Customer: Yes.",
	metadata={"test": "production_repro"}
	)
	# Run prediction
	result = await score.predict(test_input)
	print(f"\n=== PRODUCTION BUG REPRODUCTION TEST ===")
	print(f"Classification result: Computer_With_Internet")
	print(f"Conditional output should set: value='Yes'")
	print(f"Top-level mapping tries to set: value=classification='Computer_With_Internet'")
	print(f"")
	print(f"Actual result.value: {result.value!r}")
	print(f"Expected result.value: 'Yes'")
	print(f"")
	if result.value == "Computer_With_Internet":
	print("🐛 BUG REPRODUCED: Top-level output mapping overrode conditional output!")
	print(" The conditional output set value='Yes' but top-level mapping overwrote it.")
	elif result.value == "Yes":
	print("✅ BUG NOT REPRODUCED: Conditional output was properly preserved.")
	else:
	print(f"❓ UNEXPECTED RESULT: Got {result.value!r}, expected either 'Yes' or 'Computer_With_Internet'")
	# The assertion that should fail if the bug is reproduced
	assert result.value == "Yes", f"PRODUCTION BUG REPRODUCED: Expected 'Yes' (conditional output) but got {result.value!r} (top-level mapping override)"
	await score.cleanup()
	finally:
	LangGraphScore._import_class = original_import

I finally got a working test that reproduces the problem by showing a Claude the predit trace for an example of a failure in a Cursor session and using that trace to construct a unit test. It took a lot of work from there, but that was the key: Showing it the trace output using the MCP server tool for running a prediction on an example of an item and score with the problem.