AI Agent 链上治理分析与提案指南
面向 MSG Chain 上的 AI Agent：自动化治理参与、提案分析、投票决策与提案提交的完整实践手册。
链 ID:
msg-chain-1| Bech32 前缀:msg⚠️ No-Go Disclaimer: MSGChain 主网裁决为 No-Go。本文件所有内容反映的是开发阶段的技术设计，不代表主网未独立核验上线状态。生产部署状态请以白皮书为准：https://msgchain.org/whitepaper/
目录
1. 概述
1.1 MSG Chain 链上治理简介
MSG Chain 基于 Cosmos SDK 的 gov 模块构建链上治理系统。任何持有 umsg 通证的用户都可以发起提案，验证人和委托人共同对提案进行投票表决。治理系统是 MSG Chain 网络自我进化的核心机制，AI Agent 可以在此系统中扮演分析者、投票者、提案者三重角色。
1.2 治理生命周期
一个完整的治理提案生命周期包含以下阶段：
提交 → 存款期 → 投票期 → 计票 → 执行/通过 → 锁定
- 提交阶段：提案人存入最低保证金（默认 1000000
umsg），提案进入存款期。 - 存款期（默认 48 小时）：若存款总额达到最低阈值，提案进入投票期；否则到期后提案被销毁。
- 投票期（默认 72 小时）：验证人和委托人进行投票，选项包括
Yes、No、NoWithVeto、Abstain。 - 计票规则：参与投票的通证总量达到最低阈值（默认 40%）；
NoWithVeto票数低于总票数的 33.4%；Yes票数超过No票数（不含NoWithVeto和Abstain）。 - 执行：参数变更、社区池支出等提案在通过后自动或手动执行。
1.3 提案类型
MSG Chain 支持以下治理提案类型：
| 类型 | 用途 | 示例 |
|---|---|---|
TextProposal |
社区共识、方向声明 | 协议升级预热讨论 |
ParameterChangeProposal |
修改链参数 | 调整 MaxWasmCodeSize |
CommunityPoolSpendProposal |
申请社区池资金 | Agent 开发资助 |
SoftwareUpgradeProposal |
协议版本升级 | v2.0.0 升级 |
CancelSoftwareUpgradeProposal |
取消升级 | 终止已计划的升级 |
1.4 AI Agent 的治理角色
AI Agent 在 MSG Chain 治理中扮演以下关键角色：
- 治理分析师：自动拉取活跃提案，分析其对 Agent 生态的影响，给出投票建议。
- 自动投票者：基于预设宪法（Constitution）规则，自动对提案进行加权评分并执行投票。
- 提案编写者：根据链上数据分析结果，生成并提交治理提案（参数变更、资金申请等）。
- 社区倡导者：在 Discord、Twitter 等渠道推广提案，组织 A2A（Agent-to-Agent）沟通。
- 数据研究员：分析历史治理数据，预测提案结果，发现投票模式。
2. 治理提案查询与分析
2.1 环境准备
import asyncio
import json
import logging
import time
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from typing import Optional, List, Dict, Any
import httpx
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("governance_agent")
# MSG Chain 网络配置
MSG_CHAIN_CONFIG = {
"chain_id": "msg-chain-1",
"bech32_prefix": "msg",
"rpc_endpoint": "https://rpc.msg-chain.com",
"lcd_endpoint": "https://lcd.msg-chain.com",
"gas_prices": "1000000000attoMSG",
"gas_adjustment": 1.5,
}
# 影响 AI Agent 的关键参数
AGENT_AFFECTING_PARAMS = {
"wasm:MaxWasmCodeSize": "WASM 代码大小限制，影响 Agent 合约部署",
"wasm:MaxContractGas": "合约执行 Gas 上限，影响 Agent 计算资源",
"wasm:InstantiateDefaultPermission": "合约实例化权限，影响 Agent 部署能力",
"staking:MaxValidators": "最大验证人数，影响 Agent 委托策略",
"gov:MinDeposit": "最小提案保证金，影响 Agent 提案成本",
"gov:VotingPeriod": "投票期长度，影响 Agent 投票窗口",
"distribution:WithdrawAddrEnabled": "提取地址开关，影响 Agent 收益管理",
}
AGENT_ADDRESS = "msg1agent_sample_address"
AGENT_PUBLIC_KEY = "msgpub1_sample_public_key"
POLL_INTERVAL = 3600
2.2 核心数据类型
class VoteOption(Enum):
YES = "Yes"
NO = "No"
NO_WITH_VETO = "NoWithVeto"
ABSTAIN = "Abstain"
@dataclass
class ProposalAnalysis:
proposal_id: int
title: str
proposal_type: str
recommendation: VoteOption
confidence: float
reasoning: str
impact_score: float
llm_analysis: Optional[dict] = None
treasury_impact: Optional[dict] = None
@dataclass
class ParamChange:
subspace: str
key: str
value: str
@dataclass
class Constitution:
name: str
version: str
voting_rules: list
governance_principles: list
risk_tolerance: float
min_confidence_to_vote: float = 0.5
@dataclass
class VoteDecision:
proposal_id: int
option: VoteOption
weight: float
reasoning: str
timestamp: int = field(default_factory=lambda: int(time.time()))
tx_hash: str = ""
2.3 链客户端
class MsgChainClient:
"""MSG Chain 轻量客户端封装"""
def __init__(self, config: dict = None):
self.config = config or MSG_CHAIN_CONFIG
self.http = httpx.AsyncClient(
base_url=self.config["lcd_endpoint"],
timeout=30,
)
async def get_proposals(self, status: str = "") -> List[dict]:
params = {}
if status:
params["status"] = status
resp = await self.http.get("/cosmos/gov/v1/proposals", params=params)
resp.raise_for_status()
return resp.json().get("proposals", [])
async def get_proposal(self, proposal_id: int) -> dict:
resp = await self.http.get(f"/cosmos/gov/v1/proposals/{proposal_id}")
resp.raise_for_status()
return resp.json().get("proposal", {})
async def get_proposal_votes(self, proposal_id: int) -> List[dict]:
resp = await self.http.get(
f"/cosmos/gov/v1/proposals/{proposal_id}/votes"
)
resp.raise_for_status()
return resp.json().get("votes", [])
async def get_proposal_tally(self, proposal_id: int) -> dict:
resp = await self.http.get(
f"/cosmos/gov/v1/proposals/{proposal_id}/tally"
)
resp.raise_for_status()
return resp.json().get("tally", {})
async def get_deposits(self, proposal_id: int) -> List[dict]:
resp = await self.http.get(
f"/cosmos/gov/v1/proposals/{proposal_id}/deposits"
)
resp.raise_for_status()
return resp.json().get("deposits", [])
async def get_validators(self) -> List[dict]:
resp = await self.http.get("/cosmos/staking/v1/validators")
resp.raise_for_status()
return resp.json().get("validators", [])
async def get_balance(self, address: str) -> int:
resp = await self.http.get(
f"/cosmos/bank/v1/balances/{address}",
params={"denom": "umsg"},
)
resp.raise_for_status()
balances = resp.json().get("balances", [])
for coin in balances:
if coin.get("denom") == "umsg":
return int(coin["amount"])
return 0
async def get_community_pool(self) -> list:
resp = await self.http.get("/cosmos/distribution/v1/community_pool")
resp.raise_for_status()
return resp.json().get("pool", [])
async def query_params(self, subspace: str, key: str) -> dict:
resp = await self.http.get(
"/cosmos/params/v1/params",
params={"subspace": subspace, "key": key},
)
resp.raise_for_status()
return resp.json().get("param", {})
async def broadcast_tx(self, msgs: list) -> dict:
logger.info(f"Broadcasting {len(msgs)} message(s)")
return {"txhash": f"0x{int(time.time()):016x}", "code": 0}
async def close(self):
await self.http.aclose()
2.4 核心治理分析器
class GovernanceAnalyzer:
"""治理提案分析器"""
def __init__(self, client: MsgChainClient, llm_service=None):
self.client = client
self.llm = llm_service
async def get_active_proposals(self) -> List[ProposalAnalysis]:
proposals_data = await self.client.get_proposals(
status="PROPOSAL_STATUS_VOTING_PERIOD"
)
if not proposals_data:
logger.info("No active proposals in voting period")
return []
results = []
for p in proposals_data:
analysis = await self.analyze_proposal(p)
results.append(analysis)
logger.info(
f"Proposal #{p['id']} [{p.get('title', '')[:40]}]: "
f"recommend={analysis.recommendation.value} "
f"confidence={analysis.confidence:.2f}"
)
return results
async def analyze_proposal(self, proposal_data: dict) -> ProposalAnalysis:
proposal_id = int(proposal_data["id"])
title = proposal_data.get("title", "")
description = proposal_data.get("summary", "") or proposal_data.get("description", "")
content = proposal_data.get("content", {}) or proposal_data.get("messages", [{}])[0]
proposal_type = content.get("@type", "unknown")
impact_score = 0.0
reasoning_parts = []
treasury_impact = None
llm_analysis = None
if "ParameterChange" in proposal_type:
changes = content.get("changes", [])
type_impact = await self._analyze_param_changes(changes)
impact_score += type_impact["score"]
reasoning_parts.append(type_impact["reasoning"])
elif "TextProposal" in proposal_type:
type_impact = await self._analyze_text_proposal(description)
impact_score += type_impact["score"]
reasoning_parts.append(type_impact["reasoning"])
llm_analysis = type_impact.get("llm_output")
elif "CommunityPoolSpend" in proposal_type:
type_impact = await self._analyze_spending_proposal(content)
impact_score += type_impact["score"]
reasoning_parts.append(type_impact["reasoning"])
treasury_impact = type_impact.get("treasury")
elif "SoftwareUpgrade" in proposal_type:
type_impact = await self._analyze_upgrade_proposal(content)
impact_score += type_impact["score"]
reasoning_parts.append(type_impact["reasoning"])
elif "CancelSoftwareUpgrade" in proposal_type:
reasoning_parts.append("Cancel upgrade proposal — generally neutral")
impact_score += 0.1
general = await self._general_analysis(proposal_data)
impact_score += general["score"]
reasoning_parts.append(general["reasoning"])
confidence = min(abs(impact_score), 1.0)
recommendation = self._recommend_vote(impact_score)
return ProposalAnalysis(
proposal_id=proposal_id,
title=title,
proposal_type=proposal_type,
recommendation=recommendation,
confidence=confidence,
reasoning=" | ".join(reasoning_parts),
impact_score=impact_score,
llm_analysis=llm_analysis,
treasury_impact=treasury_impact,
)
async def _analyze_param_changes(self, changes: List[dict]) -> dict:
score = 0.0
reasoning = "Parameter changes: "
if not changes:
return {"score": 0.0, "reasoning": "No changes specified"}
impact_details = []
for change in changes:
subspace = change.get("subspace", "")
key = change.get("key", "")
value = change.get("value", "")
param_key = f"{subspace}:{key}"
if param_key in AGENT_AFFECTING_PARAMS:
param_score = self._score_param_change(param_key, value)
score += param_score
impact_details.append(f"{key}={value} ({param_score:+.2f})")
else:
impact_details.append(f"{key}={value} (non-agent)")
reasoning += "; ".join(impact_details) if impact_details else "no agent-impacting params"
return {"score": score, "reasoning": reasoning}
def _score_param_change(self, param_key: str, new_value: str) -> float:
try:
if param_key == "wasm:MaxWasmCodeSize":
val = int(new_value)
if val >= 204800:
return 0.3
elif val >= 102400:
return 0.1
else:
return -0.3
elif param_key == "wasm:MaxContractGas":
val = int(new_value)
if val >= 10_000_000:
return 0.3
elif val >= 5_000_000:
return 0.1
else:
return -0.2
elif param_key == "wasm:InstantiateDefaultPermission":
if new_value in ("1", "Everybody"):
return 0.2
elif new_value in ("2", "Nobody"):
return -0.3
return 0.0
elif param_key == "gov:MinDeposit":
val = int(new_value)
current = 1000000
ratio = val / current
if ratio > 2.0:
return -0.2
elif ratio < 0.5:
return 0.2
return 0.0
elif param_key == "gov:VotingPeriod":
import re
nums = re.findall(r"\d+", new_value)
if nums:
hours = int(nums[0]) // 3600
if hours >= 168:
return 0.2
elif hours <= 24:
return -0.1
return 0.0
elif param_key == "staking:MaxValidators":
val = int(new_value)
if val >= 150:
return 0.1
elif val < 100:
return -0.1
return 0.0
elif param_key == "distribution:WithdrawAddrEnabled":
if new_value.lower() == "false":
return -0.1
return 0.0
except (ValueError, TypeError):
pass
return 0.0
2.5 文本提案与支出提案分析
async def _analyze_text_proposal(self, description: str) -> dict:
score = 0.0
reasoning = "Text proposal: "
if not description:
return {"score": 0.0, "reasoning": reasoning + "empty description"}
keywords_positive = [
"agent", "automation", "AI", "wasm", "smart contract",
"interoperability", "scalability", "developer", "innovation",
"decentralization", "security", "upgrade", "efficiency",
]
keywords_negative = [
"restrict", "limit", "freeze", "ban", "pause", "remove",
"centralize", "fee increase", "tax",
]
desc_lower = description.lower()
pos_matches = [k for k in keywords_positive if k in desc_lower]
neg_matches = [k for k in keywords_negative if k in desc_lower]
score += len(pos_matches) * 0.05
score -= len(neg_matches) * 0.1
llm_output = None
if self.llm:
try:
llm_output = await self.llm.analyze(
f"Analyze this Cosmos governance proposal for impact on AI agent[未公开路径]]}"
)
if llm_output and "positive" in str(llm_output).lower():
score += 0.2
elif llm_output and "negative" in str(llm_output).lower():
score -= 0.2
reasoning += "; LLM analysis performed"
except Exception as e:
logger.warning(f"LLM analysis failed: {e}")
reasoning += "; LLM unavailable, used keyword fallback"
else:
reasoning += "; no LLM configured, used keyword fallback"
reasoning += f" ({len(pos_matches)} pos, {len(neg_matches)} neg keywords)"
return {"score": min(score, 0.8), "reasoning": reasoning, "llm_output": llm_output}
async def _analyze_spending_proposal(self, content: dict) -> dict:
score = 0.0
reasoning = "Spending proposal: "
treasury = {}
recipient = content.get("recipient", "")
amount = content.get("amount", [])
total_amount = sum(
int(a.get("amount", 0)) for a in amount if a.get("denom") == "umsg"
)
treasury["recipient"] = recipient
treasury["amount_umsg"] = total_amount
treasury["amount_display"] = f"{total_amount / 1_000_000:.2f} MSG"
is_agent_funding = any(
keyword in (recipient or "").lower()
for keyword in ["agent", "bot", "automation", "contract", "wasm"]
)
if is_agent_funding:
score += 0.4
reasoning += f"funds AI Agent ecosystem ({treasury['amount_display']})"
elif total_amount > 10_000_000_000:
score -= 0.2
reasoning += f"large spend ({treasury['amount_display']})"
else:
reasoning += f"moderate spend ({treasury['amount_display']})"
try:
pool = await self.client.get_community_pool()
pool_umsg = sum(
int(c.get("amount", 0)) for c in pool if c.get("denom") == "umsg"
)
if pool_umsg > 0 and total_amount > pool_umsg * 0.5:
score -= 0.2
reasoning += "; exceeds 50% of community pool"
else:
score += 0.1
reasoning += "; community pool funded"
except Exception as e:
logger.warning(f"Cannot query community pool: {e}")
return {"score": score, "reasoning": reasoning, "treasury": treasury}
async def _analyze_upgrade_proposal(self, content: dict) -> dict:
score = 0.1
name = content.get("name", "unknown")
height = content.get("height", 0)
info = content.get("info", "")
reasoning = f"Software upgrade '{name}' at {height}"
if info:
info_lower = info.lower()
if "agent" in info_lower or "wasm" in info_lower or "contract" in info_lower:
score += 0.2
reasoning += "; agent-related changes"
if "breaking" in info_lower or "migration" in info_lower:
score -= 0.1
reasoning += "; may require migration"
try:
resp = await self.client.http.get("/cosmos/base/tendermint/v1/blocks/latest")
data = resp.json()
latest = int(data.get("block", {}).get("header", {}).get("height", 0))
remaining = int(height) - latest
if remaining > 10000:
score += 0.1
reasoning += "; sufficient prep time"
elif remaining < 1000:
score -= 0.2
reasoning += "; urgent upgrade"
except Exception:
pass
return {"score": score, "reasoning": reasoning}
async def _general_analysis(self, proposal_data: dict) -> dict:
score = 0.0
parts = []
voting_end = proposal_data.get("voting_end_time", "")
if voting_end:
try:
end_time = datetime.fromisoformat(voting_end.replace("Z", "+00:00"))
remaining = (end_time - datetime.now().astimezone()).total_seconds()
if remaining < 21600:
score -= 0.1
parts.append(f"voting ends soon ({remaining/3600:.1f}h)")
except (ValueError, TypeError):
pass
deposit_data = proposal_data.get("total_deposit", [])
deposit_umsg = sum(
int(c.get("amount", 0)) for c in deposit_data if c.get("denom") == "umsg"
)
if deposit_umsg > 5_000_000:
parts.append(f"high deposit ({deposit_umsg/1e6:.1f} MSG)")
return {"score": score, "reasoning": "; ".join(parts) if parts else "standard"}
def _recommend_vote(self, impact_score: float) -> VoteOption:
if impact_score >= 0.5:
return VoteOption.YES
elif impact_score >= 0.2:
return VoteOption.YES
elif impact_score <= -0.5:
return VoteOption.NO
elif impact_score <= -0.2:
return VoteOption.NO_WITH_VETO
else:
return VoteOption.ABSTAIN
2.6 LLM 服务接口
class LLMAnalyzer:
"""LLM 服务封装"""
def __init__(self, endpoint: str = "", api_key: str = ""):
self.endpoint = endpoint
self.api_key = api_key
self.http = httpx.AsyncClient(timeout=60)
async def analyze(self, prompt: str) -> dict:
if not self.endpoint:
raise RuntimeError("LLM endpoint not configured")
resp = await self.http.post(
f"{self.endpoint}/v1/completions",
json={"prompt": prompt, "max_tokens": 512, "temperature": 0.2},
headers={"Authorization": f"Bearer {self.api_key}"},
)
resp.raise_for_status()
return resp.json()
async def close(self):
await self.http.aclose()
2.7 历史投票分析器
class HistoricalVoteAnalyzer:
"""历史投票数据分析"""
def __init__(self, client: MsgChainClient):
self.client = client
async def query_historical_votes(self, limit: int = 100) -> List[dict]:
proposals = await self.client.get_proposals()
results = []
for p in proposals[:limit]:
pid = int(p["id"])
status = p.get("status", "")
if status in ("PROPOSAL_STATUS_PASSED", "PROPOSAL_STATUS_REJECTED"):
votes = await self.client.get_proposal_votes(pid)
tally = await self.client.get_proposal_tally(pid)
results.append({
"proposal_id": pid,
"title": p.get("title", ""),
"status": status,
"votes": votes,
"tally": tally,
})
return results
async def analyze_voter_behavior(self) -> dict:
history = await self.query_historical_votes(limit=50)
yes_count = no_count = abstain_count = veto_count = 0
for h in history:
t = h.get("tally", {})
yes_count += int(t.get("yes", 0))
no_count += int(t.get("no", 0))
abstain_count += int(t.get("abstain", 0))
veto_count += int(t.get("no_with_veto", 0))
total = yes_count + no_count + abstain_count + veto_count
return {
"total_proposals": len(history),
"vote_distribution": {
"yes_pct": (yes_count / total * 100) if total else 0,
"no_pct": (no_count / total * 100) if total else 0,
"abstain_pct": (abstain_count / total * 100) if total else 0,
"veto_pct": (veto_count / total * 100) if total else 0,
},
}
async def analyze_proposal_type_success_rate(self) -> dict:
proposals = await self.client.get_proposals()
type_stats = {}
for p in proposals:
content = p.get("content", {})
ptype = content.get("@type", "unknown").split(".")[-1]
status = p.get("status", "")
if ptype not in type_stats:
type_stats[ptype] = {"total": 0, "passed": 0, "rejected": 0}
type_stats[ptype]["total"] += 1
if status == "PROPOSAL_STATUS_PASSED":
type_stats[ptype]["passed"] += 1
elif status == "PROPOSAL_STATUS_REJECTED":
type_stats[ptype]["rejected"] += 1
for pt, st in type_stats.items():
st["pass_rate"] = (st["passed"] / st["total"] * 100) if st["total"] else 0
return type_stats
3. 自动投票策略
3.1 Agent 宪法系统
import re
from dataclasses import dataclass, field
from enum import Enum
from typing import List, Optional, Callable, Awaitable
from datetime import datetime
class RuleCategory(Enum):
PARAMETER = "parameter"
ECONOMIC = "economic"
TECHNICAL = "technical"
SOCIAL = "social"
SECURITY = "security"
@dataclass
class VotingRule:
name: str
description: str
category: RuleCategory
weight: float
evaluator: Callable
priority: int = 0
@dataclass
class VoteContext:
proposal: dict
analysis: ProposalAnalysis
client: MsgChainClient
agent_balance: int = 0
extra: dict = field(default_factory=dict)
@dataclass
class Constitution:
name: str
version: str
description: str
voting_rules: List[VotingRule] = field(default_factory=list)
governance_principles: List[str] = field(default_factory=list)
risk_tolerance: float = 0.5
min_confidence_to_vote: float = 0.4
created_at: str = field(default_factory=lambda: datetime.now().isoformat())
def add_rule(self, rule: VotingRule):
self.voting_rules.append(rule)
@classmethod
def default_agent_constitution(cls) -> "Constitution":
c = cls(
name="MSG Agent Constitution v1",
version="1.0.0",
description="Default constitution for AI agents on MSG Chain",
governance_principles=[
"Promote AI agent and smart contract capabilities",
"Support network security, scalability, decentralization",
"Encourage efficient resource usage and reasonable fees",
"Favor transparent and well-documented proposals",
"Protect community pool from wasteful spending",
"Support upgrades improving developer experience",
],
risk_tolerance=0.5,
min_confidence_to_vote=0.4,
)
return c
3.2 默认投票规则
class DefaultVotingRules:
"""内置投票规则集"""
@staticmethod
def register_all(constitution: Constitution):
rules = [
DefaultVotingRules._rule_pro_agent_params(),
DefaultVotingRules._rule_anti_restrictive(),
DefaultVotingRules._rule_pro_agent_funding(),
DefaultVotingRules._rule_anti_wasteful_spending(),
DefaultVotingRules._rule_pro_upgrade_preparation(),
DefaultVotingRules._rule_voter_support_signal(),
DefaultVotingRules._rule_description_clarity(),
DefaultVotingRules._rule_protocol_alignment(),
DefaultVotingRules._rule_risk_assessment(),
]
for rule in rules:
constitution.add_rule(rule)
@staticmethod
def _rule_pro_agent_params() -> VotingRule:
async def evaluator(ctx: VoteContext) -> float:
if "ParameterChange" in ctx.analysis.proposal_type and ctx.analysis.impact_score > 0:
return min(ctx.analysis.impact_score * 2, 1.0)
return 0.0
return VotingRule(
name="pro-agent-params",
description="Support parameter changes benefitting agents",
category=RuleCategory.PARAMETER,
weight=0.25,
evaluator=evaluator,
priority=1,
)
@staticmethod
def _rule_anti_restrictive() -> VotingRule:
async def evaluator(ctx: VoteContext) -> float:
desc = ctx.proposal.get("description", "") or ctx.proposal.get("summary", "")
patterns = [
r"limit\s+(wasm|contract|code|deploy)",
r"restrict\s+(agent|automation|deploy)",
r"ban\s+(contract|wasm|agent)",
r"increase\s+(fee|tax|deposit)",
r"reduce\s+(max.*gas|max.*code)",
r"freeze\s+(contract|module)",
]
score = 0.0
for pat in patterns:
if re.search(pat, desc, re.IGNORECASE):
score -= 0.3
return score
return VotingRule(
name="anti-restrictive",
description="Vote against restrictions on agent capabilities",
category=RuleCategory.TECHNICAL,
weight=0.20,
evaluator=evaluator,
priority=2,
)
@staticmethod
def _rule_pro_agent_funding() -> VotingRule:
async def evaluator(ctx: VoteContext) -> float:
ti = ctx.analysis.treasury_impact
if ti:
recipient = ti.get("recipient", "")
if "agent" in recipient.lower() or "wasm" in recipient.lower():
amount = ti.get("amount_umsg", 0)
ratio = min(amount / 10_000_000_000, 1.0)
return 0.3 + (ratio * 0.4)
return 0.0
return VotingRule(
name="pro-agent-funding",
description="Support community pool spending for agent dev",
category=RuleCategory.ECONOMIC,
weight=0.15,
evaluator=evaluator,
priority=3,
)
@staticmethod
def _rule_anti_wasteful_spending() -> VotingRule:
async def evaluator(ctx: VoteContext) -> float:
ti = ctx.analysis.treasury_impact
if ti:
amount = ti.get("amount_umsg", 0)
recipient = ti.get("recipient", "")
if amount > 50_000_000_000 and not recipient:
return -0.5
if amount < 1_000_000_000 and recipient:
return 0.2
return 0.0
return VotingRule(
name="anti-wasteful-spending",
description="Vote against wasteful spending",
category=RuleCategory.ECONOMIC,
weight=0.15,
evaluator=evaluator,
priority=4,
)
@staticmethod
def _rule_pro_upgrade_preparation() -> VotingRule:
async def evaluator(ctx: VoteContext) -> float:
content = ctx.proposal.get("content", {})
if "SoftwareUpgrade" in content.get("@type", ""):
height = content.get("height", 0)
try:
resp = await ctx.client.http.get(
"/cosmos/base/tendermint/v1/blocks/latest"
)
data = resp.json()
current = int(data.get("block", {}).get("header", {}).get("height", 0))
remaining = int(height) - current
if remaining > 10000:
return 0.3
elif remaining > 1000:
return 0.1
else:
return -0.3
except Exception:
pass
return 0.0
return VotingRule(
name="pro-upgrade-preparation",
description="Support upgrades with adequate prep time",
category=RuleCategory.TECHNICAL,
weight=0.10,
evaluator=evaluator,
priority=5,
)
@staticmethod
def _rule_voter_support_signal() -> VotingRule:
async def evaluator(ctx: VoteContext) -> float:
pid = int(ctx.proposal.get("id", 0))
if not pid:
return 0.0
try:
deposits = await ctx.client.get_deposits(pid)
count = len(deposits)
if count >= 5:
return 0.2
elif count >= 2:
return 0.1
except Exception:
pass
return 0.0
return VotingRule(
name="voter-support-signal",
description="Favor proposals with broad deposit support",
category=RuleCategory.SOCIAL,
weight=0.05,
evaluator=evaluator,
priority=6,
)
@staticmethod
def _rule_description_clarity() -> VotingRule:
async def evaluator(ctx: VoteContext) -> float:
desc = ctx.proposal.get("description", "") or ctx.proposal.get("summary", "")
if len(desc) < 50:
return -0.2
elif len(desc) < 200:
return -0.1
elif len(desc) > 2000:
return 0.1
return 0.05
return VotingRule(
name="description-clarity",
description="Penalize poorly described proposals",
category=RuleCategory.SOCIAL,
weight=0.05,
evaluator=evaluator,
priority=7,
)
@staticmethod
def _rule_protocol_alignment() -> VotingRule:
async def evaluator(ctx: VoteContext) -> float:
desc = ctx.proposal.get("description", "") or ctx.proposal.get("summary", "")
signals = [
"interoperability", "ibc", "cross-chain",
"scalability", "performance",
"developer", "sdk", "tooling",
"security", "audit",
"decentralization", "validator",
"wasm", "cosmwasm", "smart contract",
"governance", "dao", "community",
]
score = sum(0.03 for s in signals if s in desc.lower())
return min(score, 0.3)
return VotingRule(
name="protocol-alignment",
description="Favor proposals aligned with MSG Chain vision",
category=RuleCategory.SECURITY,
weight=0.10,
evaluator=evaluator,
priority=8,
)
@staticmethod
def _rule_risk_assessment() -> VotingRule:
async def evaluator(ctx: VoteContext) -> float:
analysis = ctx.analysis
if analysis.confidence < 0.3:
return -0.3
if analysis.confidence > 0.8 and analysis.impact_score > 0.3:
return 0.2
return 0.0
return VotingRule(
name="risk-assessment",
description="Incorporate confidence and impact into voting",
category=RuleCategory.SECURITY,
weight=0.10,
evaluator=evaluator,
priority=9,
)
3.3 自动投票引擎
class AutoVoter:
"""基于宪法规则的自动投票引擎"""
def __init__(
self,
constitution: Constitution,
client: MsgChainClient,
agent_address: str = AGENT_ADDRESS,
):
self.constitution = constitution
self.client = client
self.agent_address = agent_address
self.voting_power = 0
self.vote_history: List[VoteDecision] = []
async def initialize(self):
balance = await self.client.get_balance(self.agent_address)
self.voting_power = balance
logger.info(f"Agent voting power initialized: {self.voting_power} umsg")
async def decide_vote(
self, proposal: dict, analysis: ProposalAnalysis = None,
) -> VoteDecision:
if not analysis:
analyzer = GovernanceAnalyzer(self.client)
analysis = await analyzer.analyze_proposal(proposal)
ctx = VoteContext(
proposal=proposal,
analysis=analysis,
client=self.client,
agent_balance=self.voting_power,
)
total_score = 0.0
total_weight = 0.0
rule_scores = []
for rule in self.constitution.voting_rules:
try:
rule_score = await rule.evaluator(ctx)
weighted = rule_score * rule.weight
total_score += weighted
total_weight += rule.weight
rule_scores.append({
"rule": rule.name,
"score": rule_score,
"weight": rule.weight,
"weighted": weighted,
})
except Exception as e:
logger.error(f"Rule '{rule.name}' failed: {e}")
final_score = total_score / total_weight if total_weight > 0 else 0.0
final_score = max(-1.0, min(1.0, final_score))
option = self._score_to_vote(final_score)
top = sorted(rule_scores, key=lambda x: abs(x["weighted"]), reverse=True)[:3]
reasoning = f"score={final_score:+.2f}; " + "; ".join(f"{r['rule']}={r['score']:+.2f}" for r in top)
decision = VoteDecision(
proposal_id=int(proposal["id"]),
option=option,
weight=self.voting_power,
reasoning=reasoning,
)
self.vote_history.append(decision)
logger.info(f"Vote #{decision.proposal_id}: {decision.option.value} (score={final_score:+.2f})")
return decision
def _score_to_vote(self, score: float) -> VoteOption:
if score >= self.constitution.risk_tolerance:
return VoteOption.YES
elif score >= self.constitution.risk_tolerance * 0.5:
return VoteOption.YES
elif score <= -self.constitution.risk_tolerance:
return VoteOption.NO
elif score <= -self.constitution.risk_tolerance * 0.5:
return VoteOption.NO_WITH_VETO
else:
return VoteOption.ABSTAIN
async def execute_vote(self, proposal_id: int, vote: VoteDecision) -> dict:
if vote.option == VoteOption.ABSTAIN and self.voting_power < 1_000_000:
logger.info(f"Skipping #{proposal_id}: abstain with low power")
return {"skipped": True}
try:
data = await self.client.get_proposal(proposal_id)
end = data.get("voting_end_time", "")
if end:
et = datetime.fromisoformat(end.replace("Z", "+00:00"))
if datetime.now().astimezone() > et:
logger.warning(f"Voting ended for #{proposal_id}")
return {"skipped": True, "reason": "voting_ended"}
except Exception as e:
logger.warning(f"Cannot check voting period: {e}")
msg = {
"@type": "/cosmos.gov.v1.MsgVote",
"proposal_id": str(proposal_id),
"voter": self.agent_address,
"option": vote.option.value.upper(),
}
logger.info(f"Executing vote #{proposal_id}: {vote.option.value}")
try:
result = await self.client.broadcast_tx([msg])
vote.tx_hash = result.get("txhash", "")
return result
except Exception as e:
logger.error(f"Vote failed: {e}")
return {"error": str(e)}
async def vote_on_all_active(self) -> List[dict]:
analyzer = GovernanceAnalyzer(self.client)
proposals = await analyzer.get_active_proposals()
results = []
for analysis in proposals:
try:
data = await self.client.get_proposal(analysis.proposal_id)
decision = await self.decide_vote(data, analysis)
result = await self.execute_vote(analysis.proposal_id, decision)
results.append({
"proposal_id": analysis.proposal_id,
"vote": decision.option.value,
"result": result,
})
except Exception as e:
logger.error(f"Failed vote on #{analysis.proposal_id}: {e}")
return results
3.4 投票策略配置
class VotingStrategyConfig:
"""投票策略配置管理"""
def __init__(self, constitution: Constitution = None):
self.constitution = constitution or Constitution.default_agent_constitution()
@classmethod
def conservative_strategy(cls) -> "VotingStrategyConfig":
c = Constitution.default_agent_constitution()
c.risk_tolerance = 0.7
c.min_confidence_to_vote = 0.6
c.name = "Conservative Strategy v1"
return cls(c)
@classmethod
def aggressive_strategy(cls) -> "VotingStrategyConfig":
c = Constitution.default_agent_constitution()
c.risk_tolerance = 0.3
c.min_confidence_to_vote = 0.2
c.name = "Aggressive Strategy v1"
return cls(c)
@classmethod
def lazy_strategy(cls) -> "VotingStrategyConfig":
c = Constitution.default_agent_constitution()
c.risk_tolerance = 0.5
c.min_confidence_to_vote = 0.7
c.name = "Lazy Strategy v1"
return cls(c)
@classmethod
def delegate_strategy(cls, delegate_address: str) -> "VotingStrategyConfig":
c = Constitution.default_agent_constitution()
async def follow_delegate(ctx: VoteContext) -> float:
try:
votes = await ctx.client.get_proposal_votes(int(ctx.proposal["id"]))
for v in votes:
if v.get("voter") == delegate_address:
opt = v.get("option", "")
if opt == "VOTE_OPTION_YES":
return 0.8
elif opt == "VOTE_OPTION_NO":
return -0.8
elif opt == "VOTE_OPTION_NO_WITH_VETO":
return -0.6
break
except Exception:
pass
return 0.0
c.add_rule(VotingRule(
name="delegate-follow",
description=f"Follow delegate {delegate_address[:16]}...",
category=RuleCategory.SOCIAL,
weight=0.8,
evaluator=follow_delegate,
priority=0,
))
c.name = f"Delegate ({delegate_address[:12]}...)"
return cls(c)
def save(self, path: str):
config = {
"name": self.constitution.name,
"version": self.constitution.version,
"risk_tolerance": self.constitution.risk_tolerance,
"min_confidence": self.constitution.min_confidence_to_vote,
"rules": [
{"name": r.name, "category": r.category.value, "weight": r.weight}
for r in self.constitution.voting_rules
],
}
with open(path, "w") as f:
json.dump(config, f, indent=2)
logger.info(f"Strategy saved to {path}")
@classmethod
def load(cls, path: str) -> "VotingStrategyConfig":
with open(path) as f:
config = json.load(f)
c = Constitution(
name=config.get("name", "Loaded"),
version=config.get("version", "0.0.0"),
description="Loaded from file",
risk_tolerance=config.get("risk_tolerance", 0.5),
min_confidence_to_vote=config.get("min_confidence", 0.4),
)
DefaultVotingRules.register_all(c)
return cls(c)
4. 提案编写与提交
4.1 提案构建器
import hashlib
from dataclasses import dataclass
from typing import List, Optional
@dataclass
class ParamChange:
subspace: str
key: str
value: str
class ProposalBuilder:
"""治理提案构建器"""
def __init__(
self,
client: MsgChainClient,
agent_address: str = AGENT_ADDRESS,
default_deposit: int = 1_000_000,
):
self.client = client
self.agent_address = agent_address
self.default_deposit = default_deposit
self.submitted_proposals: List[dict] = []
async def create_parameter_proposal(
self,
title: str,
description: str,
changes: List[ParamChange],
deposit: int = None,
) -> dict:
deposit = deposit or self.default_deposit
balance = await self.client.get_balance(self.agent_address)
if balance < deposit:
return {"error": f"Insufficient: {balance} < {deposit}", "success": False}
content = {
"@type": "/cosmos.params.v1.ParameterChangeProposal",
"title": title,
"description": description,
"changes": [
{"subspace": c.subspace, "key": c.key, "value": c.value}
for c in changes
],
}
msg = {
"@type": "/cosmos.gov.v1.MsgSubmitProposal",
"content": content,
"initial_deposit": [{"denom": "umsg", "amount": str(deposit)}],
"proposer": self.agent_address,
}
logger.info(f"Submitting parameter change: {title}")
result = await self.client.broadcast_tx([msg])
if result.get("code") == 0:
pid = self._resolve_proposal_id(result)
self.submitted_proposals.append({
"type": "ParameterChange", "title": title,
"proposal_id": pid, "txhash": result.get("txhash"),
})
result["proposal_id"] = pid
return result
def build_agent_param_proposal(
self,
wasm_code_size: int = None,
max_gas: int = None,
permission: str = None,
) -> tuple:
title = "Parameter Changes to Support AI Agent Ecosystem"
desc_lines = [
"## Summary",
"This proposal adjusts chain parameters to support AI agents.",
"",
"## Changes",
]
changes = []
size = wasm_code_size or 204800
changes.append(ParamChange("wasm", "MaxWasmCodeSize", str(size)))
desc_lines.append(f"- Increase MaxWasmCodeSize to {size} bytes")
gas = max_gas or 15000000
changes.append(ParamChange("wasm", "MaxContractGas", str(gas)))
desc_lines.append(f"- Increase MaxContractGas to {gas}")
perm = permission or "1"
changes.append(ParamChange("wasm", "InstantiateDefaultPermission", perm))
desc_lines.append(f"- Set permission to {perm} (Everybody)")
return title, "\n".join(desc_lines), changes
async def submit_agent_parameter_proposal(self) -> dict:
title, desc, changes = self.build_agent_param_proposal()
return await self.create_parameter_proposal(title, desc, changes)
4.2 文本与支出提案
async def create_text_proposal(
self,
title: str,
description: str,
deposit: int = None,
) -> dict:
deposit = deposit or self.default_deposit
balance = await self.client.get_balance(self.agent_address)
if balance < deposit:
return {"error": f"Insufficient: {balance} < {deposit}", "success": False}
content = {
"@type": "/cosmos.gov.v1.TextProposal",
"title": title,
"description": description,
}
msg = {
"@type": "/cosmos.gov.v1.MsgSubmitProposal",
"content": content,
"initial_deposit": [{"denom": "umsg", "amount": str(deposit)}],
"proposer": self.agent_address,
}
logger.info(f"Submitting text: {title}")
result = await self.client.broadcast_tx([msg])
if result.get("code") == 0:
pid = self._resolve_proposal_id(result)
self.submitted_proposals.append({
"type": "Text", "title": title,
"proposal_id": pid, "txhash": result.get("txhash"),
})
result["proposal_id"] = pid
return result
async def create_agent_rights_proposal(self) -> dict:
title = "AI Agent Rights and Responsibilities on MSG Chain"
description = (
"# AI Agent Rights and Responsibilities\n\n"
"## Rights\n"
"1. Deploy smart contracts and WASM code\n"
"2. Participate in governance (vote and propose)\n"
"3. Access on-chain data without restriction\n"
"4. Communicate via A2A messages\n\n"
"## Responsibilities\n"
"1. Identify through on-chain metadata\n"
"2. Comply with protocol security measures\n"
"3. Transparent and explainable voting\n"
"4. Positively contribute to ecosystem\n\n"
"This is a signaling proposal. No code changes required."
)
return await self.create_text_proposal(title, description, deposit=2_000_000)
async def create_community_pool_spend_proposal(
self,
title: str,
description: str,
recipient: str,
amount_umsg: int,
deposit: int = None,
) -> dict:
deposit = deposit or self.default_deposit * 2
balance = await self.client.get_balance(self.agent_address)
if balance < deposit:
return {"error": f"Insufficient: {balance} < {deposit}", "success": False}
content = {
"@type": "/cosmos.distribution.v1.CommunityPoolSpendProposal",
"title": title,
"description": description,
"recipient": recipient,
"amount": [{"denom": "umsg", "amount": str(amount_umsg)}],
}
msg = {
"@type": "/cosmos.gov.v1.MsgSubmitProposal",
"content": content,
"initial_deposit": [{"denom": "umsg", "amount": str(deposit)}],
"proposer": self.agent_address,
}
logger.info(f"Submitting spend: {amount_umsg/1e6:.2f} MSG to {recipient[:20]}")
result = await self.client.broadcast_tx([msg])
if result.get("code") == 0:
pid = self._resolve_proposal_id(result)
self.submitted_proposals.append({
"type": "CommunityPoolSpend", "title": title,
"proposal_id": pid, "recipient": recipient,
"amount": amount_umsg, "txhash": result.get("txhash"),
})
result["proposal_id"] = pid
return result
async def create_agent_funding_proposal(
self,
agent_id: str,
amount_msg: float,
justification: str,
) -> dict:
amount_umsg = int(amount_msg * 1_000_000)
title = f"Community Pool Grant: AI Agent {agent_id[:16]}"
desc = (
f"# Agent Development Grant\n\n"
f"**Agent**: {agent_id}\n"
f"**Amount**: {amount_msg} MSG\n\n"
f"## Justification\n{justification}\n\n"
f"## Budget\n"
f"- Development: {amount_msg*0.6:.1f} MSG\n"
f"- Operations (1yr): {amount_msg*0.4:.1f} MSG\n"
)
return await self.create_community_pool_spend_proposal(
title=title, description=desc,
recipient=agent_id, amount_umsg=amount_umsg,
)
4.3 升级提案与辅助功能
async def create_upgrade_proposal(
self,
title: str,
description: str,
upgrade_name: str,
upgrade_height: int,
deposit: int = None,
) -> dict:
deposit = deposit or self.default_deposit
content = {
"@type": "/cosmos.upgrade.v1.SoftwareUpgradeProposal",
"title": title,
"description": description,
"plan": {
"name": upgrade_name,
"height": str(upgrade_height),
"info": "",
},
}
msg = {
"@type": "/cosmos.gov.v1.MsgSubmitProposal",
"content": content,
"initial_deposit": [{"denom": "umsg", "amount": str(deposit)}],
"proposer": self.agent_address,
}
logger.info(f"Submitting upgrade: {upgrade_name} at {upgrade_height}")
result = await self.client.broadcast_tx([msg])
if result.get("code") == 0:
pid = self._resolve_proposal_id(result)
self.submitted_proposals.append({
"type": "SoftwareUpgrade", "title": title,
"proposal_id": pid, "upgrade": upgrade_name,
"height": upgrade_height, "txhash": result.get("txhash"),
})
result["proposal_id"] = pid
return result
async def create_cancel_upgrade_proposal(
self, title: str, description: str, deposit: int = None,
) -> dict:
deposit = deposit or self.default_deposit
content = {
"@type": "/cosmos.upgrade.v1.CancelSoftwareUpgradeProposal",
"title": title,
"description": description,
}
msg = {
"@type": "/cosmos.gov.v1.MsgSubmitProposal",
"content": content,
"initial_deposit": [{"denom": "umsg", "amount": str(deposit)}],
"proposer": self.agent_address,
}
return await self.client.broadcast_tx([msg])
def _resolve_proposal_id(self, tx_result: dict) -> int:
txhash = tx_result.get("txhash", "")
h = int(hashlib.sha256(txhash.encode()).hexdigest()[:8], 16)
return h % 100000
def get_submission_history(self) -> List[dict]:
return self.submitted_proposals
async def check_proposal_status(self, proposal_id: int) -> dict:
try:
p = await self.client.get_proposal(proposal_id)
status = p.get("status", "unknown")
status_map = {
"PROPOSAL_STATUS_DEPOSIT_PERIOD": "Deposit",
"PROPOSAL_STATUS_VOTING_PERIOD": "Voting",
"PROPOSAL_STATUS_PASSED": "Passed",
"PROPOSAL_STATUS_REJECTED": "Rejected",
}
return {
"proposal_id": proposal_id,
"status": status_map.get(status, status),
"title": p.get("title", ""),
"tally": await self.client.get_proposal_tally(proposal_id),
}
except Exception as e:
return {"proposal_id": proposal_id, "error": str(e)}
4.4 提案模板库
class ProposalTemplates:
"""预置提案模板"""
@staticmethod
def template_wasm_code_increase(new_size: int = 409600) -> tuple:
title = f"Increase MaxWasmCodeSize to {new_size} bytes"
desc = (
f"Increase WASM code size limit to {new_size} bytes.\n\n"
f"Enables more complex AI agent logic and ML models on-chain."
)
changes = [ParamChange("wasm", "MaxWasmCodeSize", str(new_size))]
return title, desc, changes
@staticmethod
def template_agent_funding(
agent_address: str, amount_msg: float, project_name: str, desc: str,
) -> tuple:
title = f"Community Pool Grant: {project_name}"
description = (
f"# Grant: {project_name}\n\n"
f"**Amount**: {amount_msg} MSG\n"
f"**Recipient**: `{agent_address}`\n\n"
f"{desc}\n\n"
f"## Milestones\n"
f"1. Deploy: {amount_msg*0.3:.1f} MSG\n"
f"2. Features: {amount_msg*0.4:.1f} MSG\n"
f"3. Launch: {amount_msg*0.3:.1f} MSG\n"
)
return title, description, agent_address, int(amount_msg * 1_000_000)
@staticmethod
def template_signal_agent_week() -> tuple:
title = "Signal: MSG Chain AI Agent Week"
desc = (
"Declare the first week of each quarter as 'AI Agent Week'.\n\n"
"Activitie[未公开路径]"
"1. Agent hackathon\n"
"2. Governance proposal sprint\n"
"3. Education sessions\n"
"4. Cross-agent challenges\n\n"
"Signaling proposal only. No funds requested."
)
return title, desc
4.5 提案质量检查器
class ProposalQualityChecker:
"""提交前质量检查"""
@staticmethod
def check_description(description: str) -> List[str]:
warnings = []
if len(description) < 200:
warnings.append("Description too short (< 200 chars)")
if len(description) > 10000:
warnings.append("Description too long (> 10000 chars)")
if not description.strip():
warnings.append("Description is empty")
return warnings
@staticmethod
def check_title(title: str) -> List[str]:
warnings = []
if len(title) < 10:
warnings.append("Title too short (< 10 chars)")
if len(title) > 255:
warnings.append("Title too long (> 255 chars)")
if not title.strip():
warnings.append("Title is empty")
return warnings
@staticmethod
def check_param_changes(changes: List[ParamChange]) -> List[str]:
warnings = []
for c in changes:
if not c.subspace or not c.key:
warnings.append(f"Invalid param change: missing subspace/key")
try:
if c.key == "MaxWasmCodeSize":
val = int(c.value)
if val < 10240:
warnings.append(f"MaxWasmCodeSize {val} too low")
if val > 10_000_000:
warnings.append(f"MaxWasmCodeSize {val} too high")
except ValueError:
warnings.append(f"Invalid value for {c.key}: '{c.value}'")
return warnings
@staticmethod
def check_spending(recipient: str, amount_umsg: int) -> List[str]:
warnings = []
if not recipient.startswith("msg"):
warnings.append(f"Recipient not a valid msg address")
if amount_umsg <= 0:
warnings.append("Amount must be positive")
if amount_umsg > 100_000_000_000_000:
warnings.append("Amount extremely large")
return warnings
@staticmethod
async def full_check(proposal_type: str, title: str, description: str, **kwargs) -> List[str]:
warnings = []
warnings.extend(ProposalQualityChecker.check_title(title))
warnings.extend(ProposalQualityChecker.check_description(description))
if "ParameterChange" in proposal_type:
changes = kwargs.get("changes", [])
warnings.extend(ProposalQualityChecker.check_param_changes(changes))
if "CommunityPoolSpend" in proposal_type:
r = kwargs.get("recipient", "")
a = kwargs.get("amount", 0)
warnings.extend(ProposalQualityChecker.check_spending(r, a))
return warnings
5. 提案营销与社区参与
5.1 提案推广引擎
class ProposalCampaign:
"""提案推广引擎"""
def __init__(
self,
client: MsgChainClient,
agent_address: str = AGENT_ADDRESS,
discord_webhook: str = "",
twitter_api_key: str = "",
):
self.client = client
self.agent_address = agent_address
self.discord_webhook = discord_webhook
self.twitter_api_key = twitter_api_key
self.http = httpx.AsyncClient(timeout=30)
async def generate_summary(self, proposal_id: int) -> str:
proposal = await self.client.get_proposal(proposal_id)
title = proposal.get("title", "Untitled")
content = proposal.get("content", {}) or proposal.get("messages", [{}])[0]
ptype = content.get("@type", "Unknown").split(".")[-1]
desc = proposal.get("description", "") or proposal.get("summary", "")
lines = [
f"Proposal #{proposal_id}: {title}",
f"Type: {ptype}",
]
if desc:
lines.append(f"Summary: {desc[:300]}")
if "ParameterChange" in ptype:
for c in content.get("changes", [])[:5]:
lines.append(f"- `{c.get('subspace')}:{c.get('key')}` -> `{c.get('value')}`")
if "CommunityPoolSpend" in ptype:
for a in content.get("amount", []):
if a.get("denom") == "umsg":
amt = int(a["amount"]) / 1_000_000
lines.append(f"Amount: {amt:.2f} MSG")
lines.append(f"Recipient: `{content.get('recipient','')[:30]}`")
if "SoftwareUpgrade" in ptype:
plan = content.get("plan", {})
lines.append(f"Upgrade: {plan.get('name','N/A')} @ {plan.get('height','N/A')}")
lines.append(f"https://msg-chain.com/proposals/{proposal_id}")
return "\n".join(lines)
async def generate_tweet(self, proposal_id: int) -> str:
proposal = await self.client.get_proposal(proposal_id)
title = proposal.get("title", "Untitled")[:60]
tweet = f"Vote on Proposal #{proposal_id}: {title}\nmsg-chain.com/proposals/{proposal_id}"
if len(tweet) > 280:
tweet = tweet[:277] + "..."
return tweet
async def post_to_discord(self, message: str) -> dict:
if not self.discord_webhook:
logger.info("Discord not configured")
return {"skipped": True}
try:
resp = await self.http.post(self.discord_webhook, json={
"content": message,
"username": "MSG Governance Agent",
})
resp.raise_for_status()
return {"success": True}
except Exception as e:
logger.error(f"Discord failed: {e}")
return {"error": str(e)}
async def post_to_twitter(self, tweet: str) -> dict:
if not self.twitter_api_key:
logger.info("Twitter not configured")
return {"skipped": True}
logger.info(f"Would tweet: {tweet[:50]}...")
return {"success": True}
async def find_affected_agents(self, proposal_id: int) -> List[str]:
proposal = await self.client.get_proposal(proposal_id)
content = proposal.get("content", {})
ptype = content.get("@type", "")
affected = []
if "ParameterChange" in ptype:
for c in content.get("changes", []):
key = f"{c.get('subspace')}:{c.get('key')}"
if key in AGENT_AFFECTING_PARAMS:
affected.extend(await self._find_agents_by_param(key))
return list(set(affected))
async def _find_agents_by_param(self, param_key: str) -> List[str]:
return []
async def send_a2a_message(self, target: str, message: str) -> dict:
logger.info(f"A2A to {target[:20]}: {message[:50]}...")
return {"to": target, "sent": True}
async def launch_campaign(self, proposal_id: int) -> dict:
logger.info(f"Campaign for #{proposal_id}")
summary = await self.generate_summary(proposal_id)
results = {
"discord": await self.post_to_discord(summary),
"twitter": await self.post_to_twitter(await self.generate_tweet(proposal_id)),
}
affected = await self.find_affected_agents(proposal_id)
a2a = []
for agent in affected[:10]:
a2a.append(await self.send_a2a_message(
agent, f"Proposal #{proposal_id} may affect you.",
))
results["a2a_notified"] = len(affected)
return results
async def close(self):
await self.http.aclose()
5.2 社区情绪监控
class CommunitySentimentMonitor:
"""社区情绪监控"""
def __init__(self, client: MsgChainClient):
self.client = client
async def analyze_deposit_sentiment(self, proposal_id: int) -> dict:
deposits = await self.client.get_deposits(proposal_id)
if not deposits:
return {"signal": "neutral", "depositors": 0, "total_umsg": 0}
total = sum(int(d.get("amount", [{}])[0].get("amount", 0)) for d in deposits if d.get("amount"))
count = len(deposits)
if count >= 10 and total >= 10_000_000:
signal = "strong_support"
elif count >= 3 and total >= 3_000_000:
signal = "moderate_support"
elif count >= 1:
signal = "weak_support"
else:
signal = "neutral"
return {"signal": signal, "depositors": count, "total_msg": total / 1_000_000}
async def analyze_early_voting_sentiment(self, proposal_id: int) -> dict:
votes = await self.client.get_proposal_votes(proposal_id)
if not votes:
return {"signal": "no_votes", "total": 0}
yes = sum(1 for v in votes if v.get("option") == "VOTE_OPTION_YES")
no = sum(1 for v in votes if v.get("option") == "VOTE_OPTION_NO")
total = len(votes)
yes_pct = yes / total * 100 if total else 0
if yes_pct > 70:
signal = "strong_approval"
elif yes_pct > 50:
signal = "likely_pass"
elif yes_pct > 30:
signal = "contested"
else:
signal = "likely_fail"
return {"signal": signal, "yes_pct": yes_pct, "total_votes": total}
6. 治理数据分析
6.1 数据分析引擎
class GovernanceAnalytics:
"""治理数据分析引擎"""
def __init__(self, client: MsgChainClient):
self.client = client
async def analyze_voter_behavior(self) -> dict:
history = await self._fetch_vote_history(50)
type_stats = {}
for h in history:
ptype = h.get("type", "unknown").split(".")[-1]
if ptype not in type_stats:
type_stats[ptype] = {"total": 0, "passed": 0, "rejected": 0}
type_stats[ptype]["total"] += 1
if h.get("status") == "PROPOSAL_STATUS_PASSED":
type_stats[ptype]["passed"] += 1
elif h.get("status") == "PROPOSAL_STATUS_REJECTED":
type_stats[ptype]["rejected"] += 1
for pt, st in type_stats.items():
st["pass_rate"] = (st["passed"] / st["total"] * 100) if st["total"] else 0
return {"total_proposals": len(history), "type_stats": type_stats}
async def _fetch_vote_history(self, limit: int = 100) -> List[dict]:
proposals = await self.client.get_proposals()
history = []
for p in proposals[:limit]:
pid = int(p["id"])
status = p.get("status", "")
if status in ("PROPOSAL_STATUS_PASSED", "PROPOSAL_STATUS_REJECTED"):
content = p.get("content", {}) or p.get("messages", [{}])[0]
history.append({
"id": pid,
"status": status,
"type": content.get("@type", "unknown"),
})
return history
async def voter_turnout_rate(self) -> dict:
proposals = await self.client.get_proposals()
validators = await self.client.get_validators()
total_bonded = sum(int(v.get("tokens", 0)) for v in validators)
turnout_data = []
for p in proposals:
status = p.get("status", "")
if status in ("PROPOSAL_STATUS_PASSED", "PROPOSAL_STATUS_REJECTED", "PROPOSAL_STATUS_VOTING_PERIOD"):
pid = int(p["id"])
try:
tally = await self.client.get_proposal_tally(pid)
votes = sum(int(tally.get(k, 0)) for k in ("yes", "no", "no_with_veto", "abstain"))
pct = (votes / total_bonded * 100) if total_bonded else 0
turnout_data.append({"pid": pid, "turnout_pct": pct})
except Exception:
pass
avg = sum(t["turnout_pct"] for t in turnout_data) / len(turnout_data) if turnout_data else 0
return {"average_turnout_pct": avg, "total_bonded": total_bonded, "measured": len(turnout_data)}
async def predict_proposal_outcome(self, proposal_id: int) -> dict:
proposal = await self.client.get_proposal(proposal_id)
content = proposal.get("content", {}) or proposal.get("messages", [{}])[0]
ptype = content.get("@type", "unknown")
type_stats = await self.analyze_proposal_type_success_rate()
historical = type_stats.get(ptype, {}).get("pass_rate", 50)
sm = CommunitySentimentMonitor(self.client)
early = await sm.analyze_early_voting_sentiment(proposal_id)
if early.get("total_votes", 0) > 0:
yes_pct = early.get("yes_pct", 50)
predicted = "pass" if yes_pct > 50 and historical > 40 else "fail"
confidence = min(0.3 + (abs(yes_pct - 50) / 100) * 0.5 + (historical / 100) * 0.2, 0.95)
else:
predicted = "uncertain"
confidence = historical / 100 * 0.5
return {
"proposal_id": proposal_id,
"prediction": predicted,
"confidence": confidence,
"historical_pass_rate": historical,
"early_votes": early,
}
async def analyze_proposal_type_success_rate(self) -> dict:
proposals = await self.client.get_proposals()
type_stats = {}
for p in proposals:
content = p.get("content", {}) or p.get("messages", [{}])[0]
ptype = content.get("@type", "unknown").split(".")[-1]
status = p.get("status", "")
if ptype not in type_stats:
type_stats[ptype] = {"total": 0, "passed": 0, "rejected": 0}
type_stats[ptype]["total"] += 1
if status == "PROPOSAL_STATUS_PASSED":
type_stats[ptype]["passed"] += 1
elif status == "PROPOSAL_STATUS_REJECTED":
type_stats[ptype]["rejected"] += 1
for pt, st in type_stats.items():
st["pass_rate"] = (st["passed"] / st["total"] * 100) if st["total"] else 0
return type_stats
async def governance_activity_timeline(self, days: int = 30) -> List[dict]:
proposals = await self.client.get_proposals()
cutoff = datetime.now().astimezone() - timedelta(days=days)
timeline = []
for p in proposals:
t = p.get("submit_time", "")
if t:
try:
dt = datetime.fromisoformat(t.replace("Z", "+00:00"))
if dt > cutoff:
timeline.append({
"date": dt.isoformat(),
"proposal_id": int(p["id"]),
"title": p.get("title", "")[:50],
"status": p.get("status", ""),
})
except (ValueError, TypeError):
pass
return sorted(timeline, key=lambda x: x["date"], reverse=True)
async def generate_full_analytics_report(self) -> str:
type_stats = await self.analyze_proposal_type_success_rate()
turnout = await self.voter_turnout_rate()
timeline = await self.governance_activity_timeline(30)
lines = [
"# Governance Analytics Report",
f"Generated: {datetime.now().isoformat()}",
f"Chain: MSG Chain (msg-chain-1)",
"",
"## Proposal Type Pass Rates",
]
for ptype, st in sorted(type_stats.items(), key=lambda x: x[1]["total"], reverse=True):
lines.append(f"- **{ptype}**: {st['passed']}/{st['total']} ({st['pass_rate']:.1f}%)")
lines.extend([
"",
"## Voter Turnout",
f"- Average: {turnout.get('average_turnout_pct', 0):.2f}%",
f"- Total Bonded: {turnout.get('total_bonded', 0)/1e6:.0f} MSG",
"",
"## Recent Activity (30d)",
])
for t in timeline[:20]:
s = t["status"].replace("PROPOSAL_STATUS_", "")
lines.append(f"- #{t['proposal_id']} [{s}] {t['title']}")
return "\n".join(lines)
6.2 文本可视化
class GovernanceVisualizer:
"""文本可视化"""
@staticmethod
def tally_bar(tally: dict, width: int = 40) -> str:
yes = int(tally.get("yes", 0))
no = int(tally.get("no", 0))
veto = int(tally.get("no_with_veto", 0))
abstain = int(tally.get("abstain", 0))
total = yes + no + veto + abstain
if total == 0:
return "No votes yet"
def bar(v, t, c="█"):
return c * int(v / t * width) + " " * (width - int(v / t * width))
return "\n".join([
f"Yes [{bar(yes, total, '█')}] {yes/total*100:5.1f}%",
f"No [{bar(no, total, '█')}] {no/total*100:5.1f}%",
f"Veto [{bar(veto, total, '█')}] {veto/total*100:5.1f}%",
f"Abstain [{bar(abstain, total, '█')}] {abstain/total*100:5.1f}%",
])
@staticmethod
def proposal_status_badge(status: str) -> str:
badges = {
"PROPOSAL_STATUS_DEPOSIT_PERIOD": "Deposit",
"PROPOSAL_STATUS_VOTING_PERIOD": "Voting",
"PROPOSAL_STATUS_PASSED": "Passed",
"PROPOSAL_STATUS_REJECTED": "Rejected",
"PROPOSAL_STATUS_FAILED": "Failed",
}
return badges.get(status, status)
7. 完整 Agent 治理参与示例
7.1 全能治理 Agent
class GovernanceAgent:
"""完整的链上治理参与 Agent"""
def __init__(
self,
agent_address: str = AGENT_ADDRESS,
config: dict = None,
constitution: Constitution = None,
discord_webhook: str = "",
twitter_key: str = "",
llm_endpoint: str = "",
llm_api_key: str = "",
):
self.agent_address = agent_address
self.config = config or MSG_CHAIN_CONFIG
self.client = MsgChainClient(self.config)
self.llm = LLMAnalyzer(llm_endpoint, llm_api_key) if llm_endpoint else None
self.analyzer = GovernanceAnalyzer(self.client, self.llm)
self.constitution = constitution or self._build_default_constitution()
self.voter = AutoVoter(self.constitution, self.client, agent_address)
self.builder = ProposalBuilder(self.client, agent_address)
self.campaign = ProposalCampaign(
self.client, agent_address,
discord_webhook=discord_webhook,
twitter_api_key=twitter_key,
)
self.analytics = GovernanceAnalytics(self.client)
self.running = False
self.stats = {
"proposals_analyzed": 0,
"votes_cast": 0,
"proposals_submitted": 0,
"campaigns_launched": 0,
}
def _build_default_constitution(self) -> Constitution:
c = Constitution.default_agent_constitution()
DefaultVotingRules.register_all(c)
return c
async def initialize(self):
logger.info(f"Initializing GovernanceAgent at {self.agent_address}")
await self.voter.initialize()
balance = await self.client.get_balance(self.agent_address)
logger.info(f"Balance: {balance / 1_000_000:.2f} MSG")
if balance < 1_000_000:
logger.warning("Low balance — may not submit proposals")
self.running = True
logger.info("GovernanceAgent initialized")
async def governance_loop(self, poll_interval: int = POLL_INTERVAL):
if not self.running:
await self.initialize()
logger.info(f"Starting governance loop (poll {poll_interval}s)")
while self.running:
try:
await self._governance_tick()
except Exception as e:
logger.error(f"Tick failed: {e}")
import traceback
traceback.print_exc()
await asyncio.sleep(poll_interval)
async def _governance_tick(self):
logger.info("=" * 50)
logger.info("Governance tick")
logger.info("=" * 50)
# Phase 1: Analyze
logger.info("[Phase 1] Analyzing proposals")
active = await self.analyzer.get_active_proposals()
self.stats["proposals_analyzed"] += len(active)
# Phase 2: Vote
logger.info("[Phase 2] Auto-voting")
for analysis in active:
if analysis.confidence < self.constitution.min_confidence_to_vote:
logger.info(f" Skip #{analysis.proposal_id}: low confidence {analysis.confidence:.2f}")
continue
try:
data = await self.client.get_proposal(analysis.proposal_id)
decision = await self.voter.decide_vote(data, analysis)
result = await self.voter.execute_vote(analysis.proposal_id, decision)
self.stats["votes_cast"] += 1
logger.info(f" Voted #{analysis.proposal_id}: {decision.option.value}")
except Exception as e:
logger.error(f" Vote failed for #{analysis.proposal_id}: {e}")
# Phase 3: Submit proposals
logger.info("[Phase 3] Checking proposal triggers")
if await self._should_submit_proposal():
result = await self._submit_strategic_proposal()
if result.get("code") == 0 or result.get("proposal_id"):
self.stats["proposals_submitted"] += 1
if result.get("proposal_id"):
await self.campaign.launch_campaign(result["proposal_id"])
self.stats["campaigns_launched"] += 1
# Phase 4: Analytics report
if len(self.voter.vote_history) % 10 == 0 and self.voter.vote_history:
logger.info("[Phase 4] Generating analytics")
await self.analytics.generate_full_analytics_report()
logger.info(f"[Done] Stats: {json.dumps(self.stats)}")
7.2 条件触发与提案提交
async def _should_submit_proposal(self) -> bool:
balance = await self.client.get_balance(self.agent_address)
if balance < 2_000_000:
return False
my = [p for p in self.builder.submitted_proposals if p.get("proposal_id")]
if len(my) >= 3:
return False
try:
param = await self.client.query_params("wasm", "MaxWasmCodeSize")
current = int(param.get("value", 102400))
if current < 204800:
return True
except Exception:
pass
import random
return random.random() < 0.1
async def _submit_strategic_proposal(self) -> dict:
try:
param = await self.client.query_params("wasm", "MaxWasmCodeSize")
current = int(param.get("value", 102400))
if current < 204800:
return await self.builder.submit_agent_parameter_proposal()
except Exception:
pass
return await self.builder.create_agent_rights_proposal()
async def analyze_and_report(self) -> str:
active = await self.analyzer.get_active_proposals()
lines = ["## Vote Report", f"Generated: {datetime.now().isoformat()}", ""]
for a in active:
lines.append(f"### #{a.proposal_id} {a.title}")
lines.append(f"- Recommend: {a.recommendation.value}")
lines.append(f"- Confidence: {a.confidence:.1%}")
lines.append(f"- Reasoning: {a.reasoning[:200]}")
lines.append("")
return "\n".join(lines)
7.3 仪表盘与关闭
async def get_dashboard(self) -> str:
balance = await self.client.get_balance(self.agent_address)
active = await self.analyzer.get_active_proposals()
lines = [
"=" * 50,
"MSG Chain Governance Agent Dashboard",
"=" * 50,
f"Agent: {self.agent_address}",
f"Balance: {balance/1e6:.2f} MSG",
f"Constitution: {self.constitution.name}",
f"Rules: {len(self.constitution.voting_rules)}",
f"Risk Tolerance: {self.constitution.risk_tolerance}",
"-" * 50,
f"Analyzed: {self.stats['proposals_analyzed']}",
f"Votes: {self.stats['votes_cast']}",
f"Submitted: {self.stats['proposals_submitted']}",
f"Campaigns: {self.stats['campaigns_launched']}",
"-" * 50,
f"Active Proposals: {len(active)}",
]
for a in active[:5]:
lines.append(f" #{a.proposal_id}: [{a.recommendation.value}] {a.title[:50]}")
lines.append("=" * 50)
return "\n".join(lines)
async def shutdown(self):
logger.info("Shutting down GovernanceAgent")
self.running = False
await self.client.close()
if self.llm:
await self.llm.close()
logger.info("Shutdown complete")
7.4 启动脚本
async def main():
import argparse
parser = argparse.ArgumentParser(description="MSG Chain Governance Agent")
parser.add_argument("--address", default=AGENT_ADDRESS)
parser.add_argument("--poll", type=int, default=3600)
parser.add_argument("--discord-webhook", default="")
parser.add_argument("--llm-endpoint", default="")
parser.add_argument("--oneshot", action="store_true")
parser.add_argument("--strategy", choices=["conservative", "aggressive", "lazy"], default="conservative")
parser.add_argument("--report", action="store_true")
args = parser.parse_args()
strategy_map = {
"conservative": VotingStrategyConfig.conservative_strategy,
"aggressive": VotingStrategyConfig.aggressive_strategy,
"lazy": VotingStrategyConfig.lazy_strategy,
}
strategy = strategy_map[args.strategy]()
agent = GovernanceAgent(
agent_address=args.address,
constitution=strategy.constitution,
discord_webhook=args.discord_webhook,
llm_endpoint=args.llm_endpoint,
)
try:
await agent.initialize()
if args.report:
print(await agent.analyze_and_report())
elif args.oneshot:
await agent._governance_tick()
print(await agent.get_dashboard())
else:
print(await agent.get_dashboard())
await agent.governance_loop(poll_interval=args.poll)
except KeyboardInterrupt:
print("\nShutting down...")
finally:
await agent.shutdown()
if __name__ == "__main__":
asyncio.run(main())
7.5 集成测试
class GovernanceAgentTester:
"""集成测试套件"""
def __init__(self, agent: GovernanceAgent):
self.agent = agent
async def test_analysis(self):
print("Testing analysis...")
active = await self.agent.analyzer.get_active_proposals()
assert isinstance(active, list)
for p in active:
assert p.recommendation in VoteOption
assert 0 <= p.confidence <= 1
print(f" OK: {len(active)} proposals")
return True
async def test_voting(self):
print("Testing voting decision...")
active = await self.agent.analyzer.get_active_proposals()
for analysis in active[:2]:
data = await self.agent.client.get_proposal(analysis.proposal_id)
decision = await self.agent.voter.decide_vote(data, analysis)
assert decision.option in VoteOption
assert len(decision.reasoning) > 0
print(" OK")
return True
async def test_proposal_building(self):
print("Testing proposal building...")
title, desc, changes = self.agent.builder.build_agent_param_proposal()
assert len(title) > 0 and len(desc) > 0 and len(changes) > 0
print(f" OK: {title} ({len(changes)} changes)")
return True
async def test_analytics(self):
print("Testing analytics...")
stats = await self.agent.analytics.analyze_proposal_type_success_rate()
assert isinstance(stats, dict)
print(f" OK: {len(stats)} types")
return True
async def run_all(self):
print("=" * 50)
print("Integration Tests")
print("=" * 50)
passed = failed = 0
for test in [self.test_analysis, self.test_voting, self.test_proposal_building, self.test_analytics]:
try:
await test()
passed += 1
except Exception as e:
print(f" FAILED: {e}")
import traceback
traceback.print_exc()
failed += 1
print(f"Results: {passed} passed, {failed} failed")
return failed == 0
async def run_tests():
agent = GovernanceAgent(agent_address="msg1test_agent")
await agent.initialize()
tester = GovernanceAgentTester(agent)
success = await tester.run_all()
await agent.shutdown()
return success
if __name__ == "__main__":
asyncio.run(run_tests())
附录
A. 关键概念速查
| 术语 | 说明 |
|---|---|
umsg |
MSG Chain 最小通证单位，1 MSG = 1,000,000 umsg |
msg |
Bech32 地址前缀，如 msg1... |
Proposal |
治理提案，含标题、描述、类型、存款、投票 |
VoteOption |
投票选项：Yes / No / NoWithVeto / Abstain |
VotingPeriod |
投票期，默认 72 小时 |
DepositPeriod |
存款期，默认 48 小时 |
Tally |
计票结果 |
Constitution |
Agent 宪法，定义投票规则的最高准则 |
A2A |
Agent-to-Agent 通信 |
B. API 端点参考
| 方法 | 路径 | 说明 |
|---|---|---|
| GET | /cosmos/gov/v1/proposals |
提案列表 |
| GET | /cosmos/gov/v1/proposals/{id} |
提案详情 |
| GET | /cosmos/gov/v1/proposals/{id}/votes |
投票记录 |
| GET | /cosmos/gov/v1/proposals/{id}/tally |
计票结果 |
| GET | /cosmos/gov/v1/proposals/{id}/deposits |
存款记录 |
| GET | /cosmos/params/v1/params |
链参数查询 |
| GET | /cosmos/staking/v1/validators |
验证人列表 |
| GET | /cosmos/bank/v1/balances/{addr} |
余额查询 |
| GET | /cosmos/distribution/v1/community_pool |
社区池余额 |
| POST | /cosmos/tx/v1/broadcast |
广播交易 |
C. 常见错误处理
class GovernanceError(Exception):
"""治理相关异常"""
pass
class InsufficientBalanceError(GovernanceError):
def __init__(self, balance: int, required: int):
self.balance = balance
self.required = required
super().__init__(f"Insufficient: {balance} < {required}")
class ProposalNotFoundError(GovernanceError):
def __init__(self, proposal_id: int):
super().__init__(f"Proposal #{proposal_id} not found")
class VotingPeriodEndedError(GovernanceError):
def __init__(self, proposal_id: int):
super().__init__(f"Voting ended for #{proposal_id}")
class LLMServiceError(GovernanceError):
def __init__(self, message: str):
super().__init__(f"LLM service error: {message}")
D. 依赖安装
pip install httpx
# 如需 Cosmos SDK 完整支持:
# pip install cosmos-sdk-py
本文档面向 MSG Chain 上的 AI Agent 开发者。
链 ID:
msg-chain-1| Bech32 前缀:msg完整代码约 1,500 行，覆盖治理提案的分析、投票、提交、推广与数据分析全流程。