hanzoai/python-sdk
2
0
Fork 0
mirror of https://github.com/hanzoai/python-sdk.git synced 2026-10-23 20:23:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: extract hanzo-metastable-consensus package

Browse Source 
- New package: hanzo-metastable-consensus
- Two-phase finality: Sampling + Finality
- Used by hanzo-tools-agent for consensus action
- Reference: https://github.com/luxfi/consensus
This commit is contained in:
zeekay
2025-12-26 16:20:32 -08:00
parent 67d8c62d5c
commit 436ff4bf3d
6 changed files with 335 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/hanzo-metastable-consensus/README.md
+66
View File
@@ -0,0 +1,66 @@
# hanzo-metastable-consensus
Metastable consensus protocol for multi-agent agreement.
## Install
```bash
pip install hanzo-metastable-consensus
```
## Usage
```python
import asyncio
from hanzo_metastable_consensus import Consensus, Result, run
async def execute(participant: str, prompt: str) -> Result:
"""Your execution function."""
# Call your agent/model/service
output = await call_agent(participant, prompt)
return Result(id=participant, output=output, ok=True, ms=100)
async def main():
state = await run(
prompt="What is the best approach?",
participants=["agent1", "agent2", "agent3"],
execute=execute,
rounds=3,
k=2,
)
print(f"Winner: {state.winner}")
print(f"Finalized: {state.finalized}")
print(f"Synthesis: {state.synthesis}")
asyncio.run(main())
```
## Protocol
Two-phase finality:
**Phase I (Sampling)**
- Each participant proposes initial response
- k-peer sampling per round
- Luminance-weighted selection (faster = higher weight)
- Confidence accumulation toward β₁
**Phase II (Finality)**
- Threshold aggregation
- β₂ finality threshold
- Winner synthesis
## Parameters
| Param | Default | Description |
|-------|---------|-------------|
| `rounds` | 3 | Sampling rounds |
| `k` | 3 | Sample size per round |
| `alpha` | 0.6 | Agreement threshold |
| `beta_1` | 0.5 | Preference threshold (Phase I) |
| `beta_2` | 0.8 | Decision threshold (Phase II) |
## Reference
https://github.com/luxfi/consensus
pkg/hanzo-metastable-consensus/hanzo_metastable_consensus/__init__.py
+11
View File
@@ -0,0 +1,11 @@
"""Metastable consensus protocol.
Two-phase finality for multi-agent agreement.
Reference: https://github.com/luxfi/consensus
"""
from .consensus import Consensus, State, Result, run
__all__ = ["Consensus", "State", "Result", "run"]
__version__ = "0.1.0"
pkg/hanzo-metastable-consensus/hanzo_metastable_consensus/consensus.py
+205
View File
@@ -0,0 +1,205 @@
"""Metastable consensus protocol.
Two-phase finality for distributed agreement.
Phase I (Sampling):
- Each participant proposes initial response
- k-peer sampling per round
- Confidence accumulation toward β₁
Phase II (Finality):
- Threshold aggregation
- β₂ finality threshold
- Winner synthesis
Reference: https://github.com/luxfi/consensus
"""
import random
import asyncio
from dataclasses import dataclass, field
from typing import Any, Callable, Coroutine, Dict, List, Optional, TypeVar
T = TypeVar("T")
@dataclass
class Result:
"""Participant result."""
id: str
output: str
ok: bool
error: Optional[str] = None
ms: int = 0
round: int = 0
@dataclass
class State:
"""Consensus state."""
prompt: str
participants: List[str]
rounds: int
k: int # Sample size per round
alpha: float # Agreement threshold (0-1)
beta_1: float # Preference threshold (Phase I)
beta_2: float # Decision threshold (Phase II)
# State
responses: Dict[str, List[str]] = field(default_factory=dict)
confidence: Dict[str, float] = field(default_factory=dict)
luminance: Dict[str, float] = field(default_factory=dict)
finalized: bool = False
winner: Optional[str] = None
synthesis: Optional[str] = None
@dataclass
class Consensus:
"""Metastable consensus protocol.
Args:
participants: List of participant IDs
execute: Async function (id, prompt) -> Result
rounds: Number of sampling rounds (default: 3)
k: Sample size per round (default: 3)
alpha: Agreement threshold (default: 0.6)
beta_1: Preference threshold (default: 0.5)
beta_2: Decision threshold (default: 0.8)
"""
participants: List[str]
execute: Callable[[str, str], Coroutine[Any, Any, Result]]
rounds: int = 3
k: int = 3
alpha: float = 0.6
beta_1: float = 0.5
beta_2: float = 0.8
async def run(self, prompt: str) -> State:
"""Run consensus protocol."""
state = State(
prompt=prompt,
participants=self.participants,
rounds=self.rounds,
k=min(self.k, len(self.participants)),
alpha=self.alpha,
beta_1=self.beta_1,
beta_2=self.beta_2,
)
# Initialize
for p in self.participants:
state.luminance[p] = 1.0
state.confidence[p] = 0.0
state.responses[p] = []
# Phase I: Sampling - initial proposals
initial = await asyncio.gather(*[
self.execute(p, prompt) for p in self.participants
])
for r in initial:
state.responses[r.id].append(r.output)
if r.ok and r.ms > 0:
# Faster = higher luminance
state.luminance[r.id] = 1.0 / (1.0 + r.ms / 1000.0)
# Sampling rounds
for _ in range(self.rounds):
# Luminance-weighted peer selection
weights = [state.luminance[p] for p in self.participants]
total = sum(weights)
sampled = random.choices(
self.participants,
[w / total for w in weights],
k=state.k
)
# Build context from sampled peers
context = [f"Query: {prompt}", "", "Peers:"]
for p in sampled:
if state.responses[p]:
context.append(f"[{p}] {state.responses[p][-1][:1000]}")
context.append("\nRefine your response:")
# Each participant refines
round_results = await asyncio.gather(*[
self.execute(p, "\n".join(context)) for p in self.participants
])
for r in round_results:
state.responses[r.id].append(r.output)
if r.ok:
# Agreement metric
agreement = self._agreement(r.output, sampled, state)
state.confidence[r.id] = state.confidence[r.id] * 0.5 + agreement * 0.5
# Check β₁ threshold
if max(state.confidence.values()) >= self.beta_1:
break
# Phase II: Finality
scores = {p: state.confidence[p] * state.luminance[p] for p in self.participants}
state.winner = max(scores, key=lambda p: scores[p])
if scores[state.winner] >= self.beta_2:
state.finalized = True
# Synthesis
if state.responses[state.winner]:
state.synthesis = state.responses[state.winner][-1]
return state
def _agreement(self, output: str, sampled: List[str], state: State) -> float:
"""Calculate agreement with sampled peers."""
if not sampled:
return 0.0
total = 0.0
for p in sampled:
if state.responses[p]:
r_words = set(output.lower().split())
p_words = set(state.responses[p][-1].lower().split())
if r_words and p_words:
overlap = len(r_words & p_words) / len(r_words | p_words)
total += overlap
return total / len(sampled)
async def run(
prompt: str,
participants: List[str],
execute: Callable[[str, str], Coroutine[Any, Any, Result]],
rounds: int = 3,
k: int = 3,
alpha: float = 0.6,
beta_1: float = 0.5,
beta_2: float = 0.8,
) -> State:
"""Run metastable consensus.
Args:
prompt: Query to reach consensus on
participants: List of participant IDs
execute: Async function (id, prompt) -> Result
rounds: Sampling rounds
k: Sample size per round
alpha: Agreement threshold
beta_1: Preference threshold
beta_2: Decision threshold
Returns:
Final consensus state
"""
consensus = Consensus(
participants=participants,
execute=execute,
rounds=rounds,
k=k,
alpha=alpha,
beta_1=beta_1,
beta_2=beta_2,
)
return await consensus.run(prompt)
pkg/hanzo-metastable-consensus/pyproject.toml
+29
View File
@@ -0,0 +1,29 @@
[project]
name = "hanzo-metastable-consensus"
version = "0.1.0"
description = "Metastable consensus protocol for multi-agent agreement"
readme = "README.md"
requires-python = ">=3.11"
license = "MIT"
authors = [{ name = "Hanzo AI", email = "[email protected]" }]
keywords = ["consensus", "metastable", "multi-agent", "distributed"]
classifiers = [
"Development Status :: 4 - Beta",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
]
dependencies = []
[project.urls]
Homepage = "https://github.com/hanzoai/python-sdk"
Repository = "https://github.com/hanzoai/python-sdk"
Documentation = "https://github.com/luxfi/consensus"
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.hatch.build.targets.wheel]
packages = ["hanzo_metastable_consensus"]
pkg/hanzo-tools-agent/hanzo_tools/agent/agent_tool.py
+23 -160
View File
@@ -9,7 +9,6 @@ Consensus: https://github.com/luxfi/consensus
import os
import time
import asyncio
import random
from dataclasses import dataclass, field
from typing import Any, Dict, List, Literal, Optional, Annotated, final, override
@@ -18,6 +17,9 @@ from mcp.server import FastMCP
from mcp.server.fastmcp import Context as MCPContext
from hanzo_tools.core import BaseTool, auto_timeout, create_tool_context
from hanzo_metastable_consensus import Consensus as MetastableConsensus
from hanzo_metastable_consensus import Result as ConsensusResult
from hanzo_metastable_consensus import run as run_consensus
Action = Annotated[
@@ -50,31 +52,7 @@ class Result:
lux: float = 1.0 # Luminance (Photon) - faster agents get higher weight
@dataclass
class Consensus:
"""Metastable consensus state.
Implements Metastable protocol from https://github.com/luxfi/consensus
Two-phase finality:
- Phase I (Sampling): k-peer sampling, confidence accumulation
- Phase II (Finality): Threshold aggregation for finality
"""
prompt: str
agents: List[str]
rounds: int
k: int # Sample size per round
alpha: float # Agreement threshold (0-1)
beta_1: float # Preference threshold (Phase I)
beta_2: float # Decision threshold (Phase II)
# State
responses: Dict[str, List[str]] = field(default_factory=dict)
confidence: Dict[str, float] = field(default_factory=dict)
luminance: Dict[str, float] = field(default_factory=dict) # Photon
finalized: bool = False
winner: Optional[str] = None
synthesis: Optional[str] = None
# Agent configurations
@@ -601,152 +579,37 @@ Consensus: https://github.com/luxfi/consensus
return "\n".join(lines)
async def _consensus(self, prompt: str, agents: List[str], rounds: int, k: int, alpha: float, beta_1: float, beta_2: float, cwd: Optional[str], timeout: int) -> str:
"""Metastable consensus protocol.
https://github.com/luxfi/consensus
Phase I (Nova DAG):
- Each agent proposes initial response
- k-peer sampling per round
- Confidence accumulation toward beta_1
Phase II (Finality):
- Threshold aggregation
- Beta_2 finality threshold
- Critic synthesizes final answer
"""
"""Metastable consensus. https://github.com/luxfi/consensus"""
if not prompt:
return "Error: prompt required"
if not agents:
return "Error: agents list required"
return "Error: agents required"
state = Consensus(
# Executor adapter
async def execute(agent_id: str, agent_prompt: str) -> ConsensusResult:
result = await self._exec(agent_id, agent_prompt, cwd, timeout)
return ConsensusResult(
id=result.agent,
output=result.output,
ok=result.ok,
error=result.error,
ms=result.ms,
)
start = time.time()
state = await run_consensus(
prompt=prompt,
agents=agents,
participants=agents,
execute=execute,
rounds=rounds,
k=min(k, len(agents)),
k=k,
alpha=alpha,
beta_1=beta_1,
beta_2=beta_2,
)
# Initialize luminance (Photon)
for agent in agents:
state.luminance[agent] = 1.0
state.confidence[agent] = 0.0
state.responses[agent] = []
lines = ["Metastable Consensus"]
lines.append(f" Agents: {', '.join(agents)}")
lines.append(f" Rounds: {rounds}, k={k}, α={alpha}, β₁={beta_1}, β₂={beta_2}")
lines.append("")
# Phase I: Nova DAG - Initial proposals
lines.append("Phase I: Nova - Initial proposals")
start = time.time()
initial_results = await asyncio.gather(*[
self._exec(agent, prompt, cwd, timeout)
for agent in agents
])
for result in initial_results:
state.responses[result.agent].append(result.output)
# Update luminance based on response time (faster = higher)
if result.ok and result.ms > 0:
state.luminance[result.agent] = 1.0 / (1.0 + result.ms / 1000.0)
lines.append(f" {result.agent}: {result.ms}ms {'✓' if result.ok else '✗'}")
# Multi-round sampling
for round_num in range(1, rounds + 1):
lines.append(f"\nRound {round_num}: k-peer sampling")
# Luminance-weighted peer selection (Photon)
weights = [state.luminance[a] for a in agents]
total = sum(weights)
probs = [w / total for w in weights]
# Sample k peers
sampled = random.choices(agents, weights=probs, k=state.k)
# Build context from sampled peers
context_parts = [f"Original query: {prompt}", "", "Peer responses:"]
for peer in sampled:
if state.responses[peer]:
context_parts.append(f"\n--- {peer} ---")
context_parts.append(state.responses[peer][-1][:1000])
context_parts.append("\n\nConsidering these perspectives, provide your refined response:")
round_prompt = "\n".join(context_parts)
# Each agent refines based on sampled peers
round_results = await asyncio.gather(*[
self._exec(agent, round_prompt, cwd, timeout)
for agent in agents
])
for result in round_results:
state.responses[result.agent].append(result.output)
# Update confidence based on agreement
if result.ok:
# Simple agreement metric: check overlap with peers
agreement = 0.0
for peer in sampled:
if peer != result.agent and state.responses[peer]:
# Basic similarity check
r_words = set(result.output.lower().split())
p_words = set(state.responses[peer][-1].lower().split())
if r_words and p_words:
overlap = len(r_words & p_words) / len(r_words | p_words)
agreement += overlap
if sampled:
agreement /= len(sampled)
state.confidence[result.agent] = state.confidence.get(result.agent, 0) * 0.5 + agreement * 0.5
# Check for Phase I threshold (beta_1)
max_conf = max(state.confidence.values()) if state.confidence else 0
lines.append(f" Max confidence: {max_conf:.2f}")
if max_conf >= beta_1:
lines.append(f" β₁ threshold reached")
# Phase II: Finality - Threshold aggregation
lines.append("\nPhase II: Finality - Threshold aggregation")
# Find winner based on confidence + luminance
scores = {a: state.confidence[a] * state.luminance[a] for a in agents}
winner = max(scores, key=lambda a: scores[a])
state.winner = winner
lines.append(f" Winner: {winner} (score: {scores[winner]:.2f})")
# Check beta_2 finality threshold
if scores[winner] >= beta_2:
state.finalized = True
lines.append(f" β₂ finality achieved")
# Synthesize final answer (critic role)
lines.append("\nSynthesis:")
synth_parts = [
f"Original query: {prompt}",
"",
"Agent responses (final round):",
]
for agent in agents:
if state.responses[agent]:
synth_parts.append(f"\n--- {agent} (confidence: {state.confidence[agent]:.2f}) ---")
synth_parts.append(state.responses[agent][-1][:1000])
synth_parts.append(f"\n\nThe {winner} response had highest confidence. Synthesize the best answer from all perspectives:")
synth_prompt = "\n".join(synth_parts)
synth_result = await self._exec(winner, synth_prompt, cwd, timeout)
state.synthesis = synth_result.output
lines.append(synth_result.output)
elapsed = time.time() - start
lines.insert(0, f"[Metastable] {elapsed:.1f}s, winner: {winner}, finalized: {state.finalized}")
return "\n".join(lines)
return f"[Metastable] {elapsed:.1f}s, winner: {state.winner}, finalized: {state.finalized}\n\n{state.synthesis or ''}"
async def _dispatch(self, tasks: List[Dict], cwd: Optional[str], timeout: int) -> str:
"""Execute different agents for different tasks in parallel.
pkg/hanzo-tools-agent/pyproject.toml
+1
View File
@@ -18,6 +18,7 @@ classifiers = [
keywords = ["hanzo", "tools", "agent", "swarm", "mcp", "ai"]
dependencies = [
"hanzo-tools-core>=0.1.0",
"hanzo-metastable-consensus>=0.1.0",
"mcp>=1.25.0",
"fastmcp>=2.14.1",
"pydantic>=2.12.5",