菜单
Improves tool-integrated reasoning by using bipartite matching to assign dense turn-level rewards, enabling credit assignment for individual tool interactions in multi-turn tasks where 4B models outperform 8B competitors.
默认使用会先检查来源的 Prompt；你也可以切换为直接命令，或下载本地副本。
决定是否安装前，请先阅读 SKILL.md，以及 SkillsMP 当前展示的配套文件。
用 Codex 或 Claude 帮你安装 复制这段 Prompt，粘贴到 Codex、Claude 或其他助手里，让它检查 Skill 页面并帮你完成安装。
直接命令不会经过审查 Prompt；运行前请先检查来源。
npx skills add https://github.com/ADu2021/skillXiv --skill matchtir-fine-grained-tool-reasoning命令会保持在同一行。复制前请横向滚动并检查完整内容。
想先保存到本地？可下载 SkillsMP 当前能够提供的文件。
基于 SOC 职业分类
正在显示 SKILL.md
| name | matchtir-fine-grained-tool-reasoning |
| title | MatchTIR: Fine-Grained Supervision for Tool-Integrated Reasoning via Bipartite Matching |
| version | 0.0.2 |
| engine | skillxiv-v0.0.2-claude-opus-4.6 |
| license | MIT |
| url | https://arxiv.org/abs/2601.10712 |
| keywords | ["tool-integration","credit-assignment","reinforcement-learning","bipartite-matching","long-horizon-reasoning"] |
| description | Improves tool-integrated reasoning by using bipartite matching to assign dense turn-level rewards, enabling credit assignment for individual tool interactions in multi-turn tasks where 4B models outperform 8B competitors. |
Enable agents to reason effectively with tools by providing fine-grained reward signals at each tool-interaction step. Rather than assigning uniform rewards across entire trajectories, use bipartite matching to align predicted tool interactions with ground-truth sequences, enabling dense per-turn credit assignment.
Match predicted tool interactions to ground-truth sequences.
# Bipartite matching for credit assignment
import numpy as np
from scipy.optimize import linear_sum_assignment
class BipartiteToolMatcher:
def __init__(self):
self.similarity_fn = self.compute_interaction_similarity
def match_sequences(self, predicted_sequence, ground_truth_sequence):
"""Match predicted to ground-truth tool interactions"""
# Create cost matrix
cost_matrix = np.zeros((len(predicted_sequence), len(ground_truth_sequence)))
for i, pred in enumerate(predicted_sequence):
for j, truth in enumerate(ground_truth_sequence):
cost_matrix[i][j] = -self.similarity_fn(pred, truth)
pred_indices, truth_indices = linear_sum_assignment(cost_matrix)
matches = []
p_idx, t_idx (pred_indices, truth_indices):
similarity = .similarity_fn(
predicted_sequence[p_idx],
ground_truth_sequence[t_idx]
)
matches.append({
: p_idx,
: t_idx,
: similarity,
: similarity > THRESHOLD
})
matches
():
tool_match = predicted[] == ground_truth[]
arg_match = .compare_arguments(predicted[], ground_truth[])
output_match = .compare_outputs(predicted[], ground_truth[])
similarity = * tool_match + * arg_match + * output_match
similarity
():
pred_args == truth_args:
.semantically_equivalent(pred_args, truth_args):
matching_keys = ( k pred_args k truth_args pred_args[k] == truth_args[k])
matching_keys / ((pred_args), (truth_args))
():
pred_output == truth_output:
:
(pred_output, (, )) (truth_output, (, )):
rel_error = (pred_output - truth_output) / ((truth_output), )
(, - rel_error)
:
Compute distinct advantages for each interaction turn.
# Turn-level advantage computation
class TurnLevelAdvantage:
def compute_turn_advantages(self, matches, trajectory_reward):
"""Assign advantage values to individual turns"""
advantages = [0.0] * len(matches)
for match in matches:
if match["is_correct"]:
# Correct interaction contributes to final reward
advantages[match["predicted_idx"]] = trajectory_reward / len(matches)
else:
# Incorrect interaction incurs penalty
advantages[match["predicted_idx"]] = -PENALTY_SCALE
# Normalize advantages
mean_adv = np.mean([a for a in advantages if a != 0.0])
std_adv = np.std([a for a in advantages if a != 0.0]) + 1e-8
normalized = [
(a - mean_adv) / std_adv if a != 0.0 else 0.0
for a in advantages
]
return normalized
def compute_dual_level_advantages(self, turn_advantages, trajectory_reward):
"""Combine turn-level and trajectory-level signals"""
# Turn-level advantages guide step selection
# Trajectory-level advantage ensures global optimization
trajectory_advantage = trajectory_reward
# Dual-level signal: local (turn) + global (trajectory)
dual_advantages = []
for turn_adv in turn_advantages:
# Balance local and global signals
combined = 0.7 * turn_adv + 0.3 * (trajectory_advantage / len(turn_advantages))
dual_advantages.append(combined)
return dual_advantages
Handle variable-length tool interaction sequences.
# Multi-turn trajectory handling
class MultiTurnTrajectory:
def __init__(self, trajectory):
self.turns = trajectory # List of tool invocations
self.length = len(trajectory)
def extract_interactions(self):
"""Extract tool interactions from trajectory"""
interactions = []
for turn in self.turns:
interaction = {
"tool": turn["tool_name"],
"args": turn["tool_args"],
"output": turn["tool_output"],
"success": turn["success"]
}
interactions.append(interaction)
return interactions
def compute_trajectory_reward(self, success_score, efficiency_score=1.0):
"""Compute overall trajectory reward"""
# Reward successful goal achievement
base_reward = 1.0 if success_score > 0.9 else 0.0
# Penalize inefficient trajectories
efficiency_penalty = 1.0 - (self.length / MAX_TRAJECTORY_LENGTH)
efficiency_adjustment = efficiency_penalty * efficiency_score
total_reward = base_reward * (1.0 + efficiency_adjustment)
return total_reward
def identify_long_horizon_challenges(self):
"""Flag trajectories with multi-step dependencies"""
challenges = {
"multi_step_dependencies": self.has_dependencies(),
"trajectory_length": self.length,
"tool_diversity": len(set(t["tool"] for t in self.turns)),
"branching_required": self.has_conditional_logic()
}
return challenges
Implement PG updates using turn-level advantages.
# Policy gradient with fine-grained rewards
class FineGrainedPolicyGradient:
def __init__(self, policy_model):
self.policy = policy_model
def update(self, trajectory, ground_truth, learning_rate=1e-3):
"""PG update with bipartite matching-based advantages"""
# 1. Match predicted to ground truth
matcher = BipartiteToolMatcher()
matches = matcher.match_sequences(
trajectory.extract_interactions(),
ground_truth
)
# 2. Compute trajectory-level reward
traj_reward = trajectory.compute_trajectory_reward(
success_score=self.compute_success(trajectory, ground_truth)
)
# 3. Compute turn-level advantages
turn_advantages = TurnLevelAdvantage().compute_turn_advantages(
matches,
traj_reward
)
# 4. Combine with trajectory-level signal
dual_advantages = TurnLevelAdvantage().compute_dual_level_advantages(
turn_advantages,
traj_reward
)
# 5. Policy gradient update for each turn
total_loss = 0.0
for turn_idx, turn in enumerate(trajectory.turns):
# Get log probability of selected action
log_prob = self.policy.get_log_prob(turn["tool_name"], turn["tool_args"])
# PG loss with advantage
loss = -log_prob * dual_advantages[turn_idx]
total_loss += loss
# Optimization step
avg_loss = total_loss / len(trajectory.turns)
self.policy.backward(avg_loss, learning_rate)
return avg_loss.item()
def compute_success(self, trajectory, ground_truth):
"""Measure how well trajectory achieved goal"""
if trajectory.extract_interactions() == ground_truth:
return 1.0
# Partial credit for mostly-correct trajectories
correct_count = sum(
1 for pred, truth in zip(trajectory.extract_interactions(), ground_truth)
if pred == truth
)
return correct_count / len(ground_truth)
Track improvements especially for complex multi-turn reasoning.
# Long-horizon evaluation
class LongHorizonEval:
def evaluate_model_size_advantage(self, model_4b, model_8b, test_set):
"""Compare small model with fine-grained rewards vs. larger baseline"""
results = {
"4b_accuracy": 0.0,
"8b_accuracy": 0.0,
"long_horizon_improvement": 0.0
}
long_horizon_tasks = [t for t in test_set if len(t["ground_truth"]) > 5]
for task in long_horizon_tasks:
# Evaluate 4B model
pred_4b = model_4b.solve(task)
correct_4b = self.is_correct(pred_4b, task)
# Evaluate 8B baseline
pred_8b = model_8b.solve(task)
correct_8b = self.is_correct(pred_8b, task)
if correct_4b and not correct_8b:
results["long_horizon_improvement"] += 1
results["4b_accuracy"] = sum(
self.is_correct(model_4b.solve(t), t) for t in long_horizon_tasks
) / len(long_horizon_tasks)
results["8b_accuracy"] = sum(
self.is_correct(model_8b.solve(t), t) for t in long_horizon_tasks
) / len(long_horizon_tasks)
return results
def is_correct(self, prediction, task):
"""Check if prediction matches ground truth"""
return prediction == task["ground_truth"]