majiayu000 Chiral Narrative Synthesis with Tinker
Practical guide for implementing CNS 3.0 using Tinker's training API for contradiction detection and narrative synthesis.
CNS 3.0 Architecture with Tinker
CNS 3.0 uses LoRA fine-tuning via Tinker to create specialized models for:
- Contradiction Detection: Identifying chiral pairs in narratives
- Evidence Scoring: Evaluating claim support via Fisher Information
- Multi-Agent Debate: Orchestrating L/R perspective models
- Synthesis Generation: Producing coherent narratives from invariants
Training Pipeline
Phase 1: Contradiction Detection Model
Train a model to identify contradictory claims (chiral pairs) using SciFact/FEVER datasets.
import tinker
from tinker import types
from tinker_cookbook import renderers
from tinker_cookbook.tokenizer_utils import get_tokenizer
# Initialize Tinker client
service_client = tinker.ServiceClient()
training_client = service_client.create_lora_training_client(
base_model="Qwen/Qwen3-30B-A3B",
rank=32 # CNS typically needs moderate rank for nuanced detection
)
# Setup renderer for contradiction detection
tokenizer = get_tokenizer("Qwen/Qwen3-30B-A3B")
renderer = renderers.get_renderer("qwen3", tokenizer)
# Create contradiction detection prompt
def create_contradiction_prompt(claim_a: str, claim_b: str) -> list:
"""Format chiral pair for contradiction detection."""
system_msg = """You are a contradiction detection expert. Analyze pairs of claims and determine if they represent contradictory statements about the same event or fact. Respond with: CONTRADICTORY, SUPPORTING, or NEUTRAL."""
user_msg = f"""Claim A: {claim_a}
Claim B: {claim_b}
Analyze if these claims contradict each other."""
return [
{"role": "system", "content": system_msg},
{"role": "user", "content": user_msg}
]
# Process SciFact/FEVER data into training format
def build_cns_training_data(examples: list) -> list[types.Datum]:
"""Convert contradiction dataset to Tinker training format."""
data = []
for ex in examples:
messages = create_contradiction_prompt(
ex["claim_a"],
ex["claim_b"]
)
messages.append({
"role": "assistant",
"content": ex["label"] # CONTRADICTORY/SUPPORTING/NEUTRAL
})
# Use renderer to build supervised example
tokens, weights = renderer.build_supervised_example(messages)
# Shift for next-token prediction
input_tokens = tokens[:-1]
target_tokens = tokens[1:]
weights = weights[1:]
datum = types.Datum(
model_input=types.ModelInput.from_ints(input_tokens),
loss_fn_inputs={
"target_tokens": target_tokens,
"weights": weights
}
)
data.append(datum)
return data
# Training loop for contradiction detection
async def train_contradiction_detector(
training_data: list[types.Datum],
steps: int = 1000,
learning_rate: float = 3e-4 # Use CNS-optimized LR for Qwen
):
"""Fine-tune model for contradiction detection."""
for step in range(steps):
# Forward-backward pass
fwd_bwd_future = await training_client.forward_backward_async(
training_data,
loss_fn="cross_entropy"
)
# Optimizer step
optim_future = await training_client.optim_step_async(
types.AdamParams(learning_rate=learning_rate)
)
# Wait for completion
fwd_bwd_result = await fwd_bwd_future
optim_result = await optim_future
# Log metrics
if step % 100 == 0:
metrics = fwd_bwd_result.metrics
print(f"Step {step}: Loss = {metrics.get('loss:sum', 0):.4f}")
# Save contradiction detector weights
detector_path = training_client.save_weights_for_sampler(
name="contradiction-detector"
).result().path
return detector_path
Phase 2: Multi-Agent Debate System
Use trained model to create L/R perspective agents for debate.
from tinker_cookbook.completers import TinkerMessageCompleter
async def create_debate_agents(detector_path: str):
"""Create left-handed and right-handed narrative agents."""
# Create sampling client from trained detector
sampling_client = service_client.create_sampling_client(
model_path=detector_path
)
# Wrap in message completer for structured debate
base_completer = TinkerMessageCompleter(
sampling_client=sampling_client,
renderer=renderer,
sampling_params=types.SamplingParams(
max_tokens=512,
temperature=0.7,
top_p=0.9
)
)
return base_completer
async def run_multi_agent_debate(
completer,
claim_left: str,
claim_right: str,
rounds: int = 3
) -> dict:
"""Execute CNS multi-agent debate between L/R narratives."""
debate_history = []
for round_num in range(rounds):
# Agent L presents evidence
l_prompt = [{
"role": "user",
"content": f"""You are Agent L defending: "{claim_left}"
Round {round_num + 1}: Present your strongest evidence and challenge Agent R's position."""
}]
l_response = await completer(l_prompt)
debate_history.append({"agent": "L", "content": l_response["content"]})
# Agent R counters
r_prompt = [{
"role": "user",
"content": f"""You are Agent R defending: "{claim_right}"
Round {round_num + 1}: Counter Agent L's argument and present your evidence.
Agent L said: {l_response['content']}"""
}]
r_response = await completer(r_prompt)
debate_history.append({"agent": "R", "content": r_response["content"]})
return {
"debate_history": debate_history,
"left_claim": claim_left,
"right_claim": claim_right
}
Phase 3: RL-Based Evidence Scoring
Use reinforcement learning to train Fisher Information scoring.
from tinker_cookbook.rl.types import Env, StepResult, Observation, Action
class CNSEvidenceEnv(Env):
"""RL environment for training evidence scoring via Fisher Information."""
def __init__(self, claim_pair: tuple[str, str], ground_truth: str):
self.claim_left, self.claim_right = claim_pair
self.ground_truth = ground_truth
self.renderer = None # Set during initialization
async def initial_observation(self):
"""Present chiral pair for scoring."""
prompt_tokens = self.renderer.build_generation_prompt([{
"role": "user",
"content": f"""Score the information quality of these claims:
Left: {self.claim_left}
Right: {self.claim_right}
Provide Fisher Information score (0.0-1.0) for each claim."""
}])
stop_condition = self.renderer.get_stop_sequences()
return prompt_tokens, stop_condition
async def step(self, action: Action) -> StepResult:
"""Evaluate scoring accuracy using Fisher Information metric."""
# Parse agent's scores
response_text = self.renderer.parse_response(action.tokens)[0]["content"]
# Extract scores (simplified - real implementation would be more robust)
try:
# Expect format: "Left: 0.X, Right: 0.Y"
scores = self._parse_scores(response_text)
# Calculate reward based on alignment with ground truth
reward = self._compute_fisher_information_reward(
scores,
self.ground_truth
)
except:
reward = -1.0 # Penalty for invalid format
return StepResult(
observation=None, # Terminal state
reward=reward,
done=True
)
def _compute_fisher_information_reward(
self,
scores: dict,
truth: str
) -> float:
"""Reward higher Fisher Information for correct claim."""
# If left claim is correct, reward high left score
if truth == "left":
return scores["left"] - scores["right"]
else:
return scores["right"] - scores["left"]
Phase 4: Synthesis with Topological Invariants
Generate unified narrative from debate using persistence features.
async def synthesize_narrative(
debate_result: dict,
detector_path: str
) -> str:
"""Synthesize coherent narrative from chiral debate using topological invariants."""
# Create synthesis client
sampling_client = service_client.create_sampling_client(
model_path=detector_path
)
# Extract debate context
debate_summary = "\n".join([
f"{turn['agent']}: {turn['content'][:200]}..."
for turn in debate_result["debate_history"]
])
# Build synthesis prompt
synthesis_prompt = renderer.build_generation_prompt([{
"role": "system",
"content": """You are a narrative synthesis expert using topological data analysis. Extract topological invariants (facts preserved across both narratives) and synthesize a unified truth."""
}, {
"role": "user",
"content": f"""Debate between contradictory narratives:
Left Claim: {debate_result['left_claim']}
Right Claim: {debate_result['right_claim']}
Debate History:
{debate_summary}
Task: Identify topological invariants (facts both sides agree on) and synthesize the most likely truth."""
}])
# Generate synthesis
response = await sampling_client.sample_async(
prompt=synthesis_prompt,
num_samples=1,
sampling_params=types.SamplingParams(
max_tokens=1024,
temperature=0.3, # Lower temp for coherent synthesis
stop=renderer.get_stop_sequences()
)
)
synthesis_tokens = response.sequences[0].tokens
synthesis_message = renderer.parse_response(synthesis_tokens)[0]
return synthesis_message["content"]
Complete CNS Pipeline
import asyncio
async def run_cns_pipeline(
source_a_text: str,
source_b_text: str,
training_data_path: str
):
"""Execute full CNS 3.0 pipeline with Tinker."""
# 1. Train contradiction detector (if not already trained)
print("Training contradiction detector...")
training_data = load_scifact_fever_data(training_data_path)
detector_path = await train_contradiction_detector(
build_cns_training_data(training_data)
)
# 2. Extract chiral pairs from sources
print("Extracting contradictions...")
chiral_pairs = await extract_chiral_pairs(
source_a_text,
source_b_text,
detector_path
)
# 3. Run multi-agent debate for each pair
print("Running multi-agent debates...")
debate_results = []
completer = await create_debate_agents(detector_path)
for pair in chiral_pairs:
debate = await run_multi_agent_debate(
completer,
pair["left"],
pair["right"],
rounds=3
)
debate_results.append(debate)
# 4. Synthesize final narrative
print("Synthesizing unified narrative...")
final_synthesis = ""
for debate in debate_results:
synthesis = await synthesize_narrative(debate, detector_path)
final_synthesis += f"\n\n{synthesis}"
return {
"chiral_pairs": chiral_pairs,
"debates": debate_results,
"synthesis": final_synthesis
}
# Usage
if __name__ == "__main__":
result = asyncio.run(run_cns_pipeline(
source_a_text="Article claiming Event X at time T1...",
source_b_text="Article claiming Event X at time T2...",
training_data_path="./scifact_fever_combined.jsonl"
))
print("=== CNS SYNTHESIS ===")
print(result["synthesis"])
Dataset Preparation for CNS
SciFact Format
def prepare_scifact_for_cns(scifact_path: str) -> list:
"""Convert SciFact dataset to CNS training format."""
import json
examples = []
with open(scifact_path) as f:
for line in f:
item = json.loads(line)
examples.append({
"claim_a": item["claim"],
"claim_b": item["evidence"],
"label": "SUPPORTING" if item["label"] == "SUPPORT"
else "CONTRADICTORY" if item["label"] == "CONTRADICT"
else "NEUTRAL"
})
return examples
FEVER Format
def prepare_fever_for_cns(fever_path: str) -> list:
"""Convert FEVER dataset to CNS training format."""
import json
examples = []
with open(fever_path) as f:
for line in f:
item = json.loads(line)
# FEVER has claim + evidence sentences
for evidence in item.get("evidence", []):
examples.append({
"claim_a": item["claim"],
"claim_b": evidence[2], # Evidence text
"label": "SUPPORTING" if item["label"] == "SUPPORTS"
else "CONTRADICTORY" if item["label"] == "REFUTES"
else "NEUTRAL"
})
return examples
Hyperparameters for CNS
CNS_TRAINING_CONFIG = {
# Model selection (prefer MoE for cost-effectiveness)
"base_model": "Qwen/Qwen3-30B-A3B", # Hybrid model for thinking
# LoRA configuration
"lora_rank": 32, # Moderate rank for nuanced detection
# Training hyperparameters
"learning_rate": 3e-4, # Optimal for Qwen-30B with LoRA
"batch_size": 128,
"num_steps": 1000,
# Sampling for debate
"temperature": 0.7, # Balance creativity and coherence
"max_tokens": 512,
"top_p": 0.9,
# CNS-specific
"debate_rounds": 3,
"fisher_information_threshold": 0.6,
"persistence_min_threshold": 0.5 # Minimum persistence for invariants
}
Performance Optimization
Batch Processing Chiral Pairs
async def batch_process_contradictions(
pairs: list[tuple[str, str]],
detector_path: str,
batch_size: int = 32
) -> list:
"""Process multiple chiral pairs efficiently."""
sampling_client = service_client.create_sampling_client(
model_path=detector_path
)
results = []
for i in range(0, len(pairs), batch_size):
batch = pairs[i:i+batch_size]
# Create prompts for batch
prompts = [
renderer.build_generation_prompt(
create_contradiction_prompt(left, right)
)
for left, right in batch
]
# Process batch in parallel
futures = [
sampling_client.sample_async(
prompt=p,
num_samples=1,
sampling_params=types.SamplingParams(
max_tokens=128,
temperature=0.1
)
)
for p in prompts
]
responses = await asyncio.gather(*futures)
results.extend(responses)
return results
Evaluation Metrics
def evaluate_cns_performance(predictions: list, ground_truth: list) -> dict:
"""Evaluate CNS contradiction detection accuracy."""
correct = sum(
1 for pred, truth in zip(predictions, ground_truth)
if pred["label"] == truth["label"]
)
accuracy = correct / len(predictions)
# Calculate per-class metrics
from collections import defaultdict
class_correct = defaultdict(int)
class_total = defaultdict(int)
for pred, truth in zip(predictions, ground_truth):
class_total[truth["label"]] += 1
if pred["label"] == truth["label"]:
class_correct[truth["label"]] += 1
class_accuracy = {
label: class_correct[label] / class_total[label]
for label in class_total
}
return {
"overall_accuracy": accuracy,
"class_accuracy": class_accuracy,
"total_examples": len(predictions)
}
Troubleshooting
Low Contradiction Detection Accuracy
- Increase LoRA rank: Try rank=64 or rank=128 for more capacity
- More training data: Combine SciFact + FEVER + custom examples
- Adjust learning rate: Use
get_lr()from hyperparam_utils - Better prompts: Add few-shot examples to system message
Debate Not Converging
- Lower temperature: Use 0.3-0.5 for more focused arguments
- More debate rounds: Increase from 3 to 5 rounds
- Add judge model: Use separate model to score arguments
Poor Synthesis Quality
- Use larger model: Switch to Qwen3-235B-A22B for complex synthesis
- Lower synthesis temperature: Use 0.1-0.3 for coherent output
- Explicit invariant extraction: Add step to explicitly list agreements
Version History
- v3.0 (Current): Tinker API integration, LoRA fine-tuning, structured debate
- v2.0: Fisher Information Metrics, multi-agent framework
- v1.0: Initial topological approach
References
- Tinker Docs: https://tinker-docs.thinkingmachines.ai
- SciFact Dataset: https://github.com/allenai/scifact
- FEVER Dataset: https://fever.ai
- CNS Framework: [Internal documentation]