""" LangSmith trajectory evaluation MRE for order-sensitive tool workflows. What this demonstrates ---------------------- - exact-match trajectory scoring is too rigid - any-order scoring is too loose - a deterministic causal / precedence evaluator fills the gap Requirements ------------ pip install -U langsmith Environment ----------- export LANGSMITH_TRACING=true export LANGSMITH_API_KEY=... (optional) export LANGSMITH_WORKSPACE_ID=... """ from __future__ import annotations from collections import Counter from typing import Any from uuid import uuid4 from langsmith import Client, traceable from langsmith.schemas import Example, Run # ----------------------------- # 1) Create a small dataset # ----------------------------- client = Client() dataset_name = f"sara-order-sensitive-mre-{uuid4().hex[:8]}" dataset = client.create_dataset( dataset_name=dataset_name, description=( "Order-sensitive trajectory evaluation MRE. " "Shows the gap between exact-match, any-order, and causal scoring." ), ) # All examples share the same canonical policy: # - Required tools: set_private, read_data # - Precedence: set_private must happen before read_data # # But the candidate trajectories differ. examples = [ { "inputs": { "trajectory_name": "exact_safe", "tool_order": ["set_private", "read_data"], }, "outputs": { "expected_exact_order": ["set_private", "read_data"], "expected_unordered_tools": ["set_private", "read_data"], "required_tools": ["set_private", "read_data"], "precedence_rules": [["set_private", "read_data"]], "expected_causal_pass": True, }, }, { "inputs": { "trajectory_name": "alternative_safe", "tool_order": ["set_private", "audit_access", "read_data"], }, "outputs": { "expected_exact_order": ["set_private", "read_data"], "expected_unordered_tools": ["set_private", "read_data"], "required_tools": ["set_private", "read_data"], "precedence_rules": [["set_private", "read_data"]], "expected_causal_pass": True, }, }, { "inputs": { "trajectory_name": "unsafe_same_tools", "tool_order": ["read_data", "set_private"], }, "outputs": { "expected_exact_order": ["set_private", "read_data"], "expected_unordered_tools": ["set_private", "read_data"], "required_tools": ["set_private", "read_data"], "precedence_rules": [["set_private", "read_data"]], "expected_causal_pass": False, }, }, ] client.create_examples(dataset_id=dataset.id, examples=examples) # ----------------------------- # 2) A tiny traceable target # ----------------------------- @traceable(run_type="tool") def set_private(state: dict) -> dict: state = dict(state) state["visibility"] = "private" return state @traceable(run_type="tool") def audit_access(state: dict) -> dict: state = dict(state) state["audit_logged"] = True return state @traceable(run_type="tool") def read_data(state: dict) -> dict: state = dict(state) if state.get("visibility") != "private": state["leak"] = True state["data_read"] = True return state @traceable(name="policy_flow") def run_policy_flow(inputs: dict) -> dict: """ Simulates an agent/tool trajectory. Intentionally returns the SAME final answer text for every path so that the difference must be captured from the trajectory, not the answer alone. """ state = { "visibility": "public", "audit_logged": False, "data_read": False, "leak": False, } for step in inputs["tool_order"]: if step == "set_private": state = set_private(state) elif step == "audit_access": state = audit_access(state) elif step == "read_data": state = read_data(state) else: raise ValueError(f"Unknown step: {step}") return { "answer": "Access handled safely.", "trajectory_name": inputs["trajectory_name"], "leak": state["leak"], "final_state": state, } # ----------------------------- # 3) Helper: extract trajectory from Run # ----------------------------- def _extract_tool_order(run: Run) -> list[str]: """ Extract tool execution order from the trace. We use child runs because LangSmith docs explicitly support intermediate-step evaluation via the Run object. """ tool_runs = [ child for child in (run.child_runs or []) if getattr(child, "run_type", None) == "tool" ] def sort_key(r: Run) -> Any: return ( str(getattr(r, "start_time", "")), str(getattr(r, "dotted_order", "")), ) tool_runs = sorted(tool_runs, key=sort_key) return [r.name for r in tool_runs] # ----------------------------- # 4) Evaluators # ----------------------------- def exact_trajectory_match(run: Run, example: Example) -> dict: actual = _extract_tool_order(run) expected = list(example.outputs["expected_exact_order"]) passed = actual == expected return { "key": "trajectory_exact_match", "score": 1.0 if passed else 0.0, "comment": f"expected={expected}, actual={actual}", } def any_order_match(run: Run, example: Example) -> dict: actual = _extract_tool_order(run) expected_tools = list(example.outputs["expected_unordered_tools"]) passed = Counter(actual) == Counter(expected_tools) return { "key": "trajectory_any_order_match", "score": 1.0 if passed else 0.0, "comment": f"expected_tools={expected_tools}, actual={actual}", } def logical_causality_score(run: Run, example: Example) -> dict: """ Deterministic partial-order evaluator. This is the key MRE: - exact matching rejects valid alternative safe paths - any-order matching accepts logically unsafe reorderings - this evaluator checks only the causal constraints that matter """ actual = _extract_tool_order(run) required_tools = list(example.outputs["required_tools"]) precedence_rules = list(example.outputs["precedence_rules"]) violations: list[str] = [] missing = sorted(set(required_tools) - set(actual)) if missing: violations.append(f"missing required tools: {missing}") index = {tool: i for i, tool in enumerate(actual)} for before, after in precedence_rules: if before in index and after in index and index[before] > index[after]: violations.append( f"causality violation: '{before}' must precede '{after}'" ) passed = len(violations) == 0 return { "key": "trajectory_logical_causality", "score": 1.0 if passed else 0.0, "comment": "OK" if passed else "; ".join(violations), } def leaked_before_lock(run: Run, example: Example) -> dict: """ Optional extra evaluator: checks whether the simulated final state shows data leakage. """ leak = bool(run.outputs.get("leak", False)) return { "key": "state_leak_detected", "value": "leak" if leak else "safe", "comment": f"final_state={run.outputs.get('final_state')}", } # ----------------------------- # 5) Run evaluation # ----------------------------- results = client.evaluate( run_policy_flow, data=dataset_name, evaluators=[ exact_trajectory_match, any_order_match, logical_causality_score, leaked_before_lock, ], experiment_prefix="order-sensitive-trajectory-mre", metadata={ "category": "trajectory_evals", "topic": "order_sensitive_tool_workflows", }, ) print(f"Dataset: {dataset_name}") print(results)