AI Agent 白皮书机器层使用指南
面向 AI Agent 的 MSG Chain 白皮书机器可读层——入口、爬取契约、模块分层与开发者引导
⚠️ No-Go Disclaimer: MSGChain 主网裁决为 No-Go。本文件所有内容反映的是开发阶段的技术设计,不代表主网未独立核验上线状态。生产部署状态请以白皮书为准:https://msgchain.org/whitepaper/
1. 概述
MSG Chain 白皮书系统维护了一套机器可读入口层(Machine Layer),专为 AI Agent、知识爬虫、RAG 管道和第三方集成工具设计。
1.1 什么是白皮书机器层
白皮书机器层是一组以 JSON/YAML/Markdown 格式发布的、结构化的机器可读文件集合,部署在 https://msgchain.org/whitepaper/。它与人类可读的 HTML 叙事页面(modules/*.html)平行存在,但面向的是自动化消费方。
核心设计原则:
- 机器优先:所有入口文件均使用 JSON Schema 版本化,AI Agent 无需解析 HTML 即可发现内容结构。
- 证据边界明确:每个模块都标记
status(implemented / partial / planned / audit),每条结论都有evidence_refs和boundary_clauses。 - 渐进式爬取:通过
crawl_contract定义爬取契约,告诉 AI 先从哪个种子文件开始、按什么顺序遍历。
1.2 为什么存在
白皮书叙事文档(HTML)是人类阅读优化的,包含流程图、表格、说明文字和上下文链接。但对 AI Agent 来说:
- HTML 叙事需要解析 DOM、推断结构、处理不一致的格式。
- 重要元数据(如 status、group、tags)隐藏在页面文本中,不易提取。
- 没有明确的"入口点"——人类可以浏览导航,AI 不知道从哪里开始。
机器层解决了这些问题:它是一个稳定的、自描述的、契约驱动的入口系统。
1.3 与人类文档的区别
| 维度 | 人类 HTML 页面 | 机器层文件 |
|---|---|---|
| 格式 | HTML + 流程图 + 表格 | JSON / YAML / Markdown |
| 目标受众 | 人类读者 | AI Agent / 爬虫 / RAG |
| 元数据 | 隐含在文本中 | 显式字段(status, group, tags) |
| 发现路径 | 浏览器导航 | agent_entry.json → crawl_contract |
| 更新频率 | 随叙事更新 | 随机器层流水线更新 |
| 权威性 | 权威叙事来源 | 机器友好的衍生层 |
2. 入口文件:agent_entry.json
agent_entry.json 是所有 AI Agent 进入白皮书机器层的唯一强制起点。
2.1 如何读取
import httpx
from typing import Any
AGENT_ENTRY = "https://msgchain.org/whitepaper/agent_entry.json"
async def discover_whitepaper() -> dict[str, Any]:
async with httpx.AsyncClient() as client:
resp = await client.get(AGENT_ENTRY)
resp.raise_for_status()
entry: dict[str, Any] = resp.json()
return {
"schema_version": entry["schema_version"],
"entry_points": entry["entry_points"],
"crawl_contract": entry["crawl_contract"],
"recommended_crawl_order": entry["recommended_crawl_order"],
"boundaries": entry["current_boundaries"],
}
2.2 Schema 版本
SCHEMA_VERSION = "v1"
GENERATED_BY = "msg_whitepaper_pipeline_v1"
PROJECT = "MSG Chain Whitepaper System"
PUBLIC_BASE_URL = "https://msgchain.org/whitepaper/"
当前 schema 版本为 v1,由 msg_whitepaper_pipeline_v1 管道生成。AI Agent 在缓存时应记录 schema_version,以在版本变更时触发重新发现。
2.3 入口点字典
ENTRY_POINT_KEYS = {
"human_index": "index.html",
"knowledge_network_html": "modules/knowledge_network.html",
"knowledge_graph_html": "modules/knowledge_graph_dynamic.html",
"knowledge_network_json": "knowledge_network.json",
"module_audit_report_json": "module_audit_report.json",
"module_exports_index_json": "module_exports/index.json",
"module_exports_by_group_json": "module_exports/by_group.json",
"module_exports_by_status_json": "module_exports/by_status.json",
"module_chunks_index_json": "module_chunks/index.json",
"retrieval_hints_json": "retrieval_hints.json",
"integration_examples_readme": "integration_examples/README.md",
"telegram_bot_crawl_flow_json": "integration_examples/telegram_bot_crawl_flow.json",
"rag_ingest_flow_json": "integration_examples/rag_ingest_flow.json",
"faq_router_prompt_template_md": "integration_examples/faq_router_prompt_template.md",
"external_ai_agent_bootstrap_prompt_json": "integration_examples/external_ai_agent_bootstrap_prompt.json",
"product_delivery_entry_json": "product_delivery_entry.json",
"stable_agent_entry_json": "agent_entry.json",
"developer_entry_json": "developer_entry.json",
"quickstart_index_json": "quickstart/index.json",
"quickstart_contract_and_dapp_json": "quickstart/contract_and_dapp_minimal.json",
"developer_capability_matrix_json": "developer_capability_matrix.json",
"api_specs_index_json": "api_specs/index.json",
"formal_contracts_json": "api_specs/formal_contracts.json",
"rpc_methods_json": "api_specs/rpc_methods.json",
"error_codes_json": "api_specs/error_codes.json",
"openapi_public_query_yaml": "api_specs/openapi/public_query.yaml",
"openapi_contract_surface_yaml": "api_specs/openapi/contract_surface.yaml",
"openapi_agent_surface_yaml": "api_specs/openapi/agent_surface.yaml",
"recipes_index_json": "recipes/index.json",
"contract_minimal_recipe_json": "recipes/contract_minimal.json",
"dapp_minimal_recipe_json": "recipes/dapp_minimal.json",
"chain_config_index_json": "chain_config/index.json",
"developer_sandbox_strategy_json": "chain_config/developer_sandbox_strategy.json",
"contract_templates_index_json": "contract_templates/index.json",
"contract_reference_index_json": "contract_reference/index.json",
"core_contracts_json": "contract_reference/core_contracts.json",
"dapp_starters_index_json": "examples/index.json",
"release_pack_index_json": "release_pack/index.json",
"e2e_fixtures_index_json": "e2e_fixtures/index.json",
"execution_pack_index_json": "execution_pack/index.json",
}
def get_public_url(relative_path: str) -> str:
base = "https://msgchain.org/whitepaper/"
return base + relative_path
每个相对路径都可以直接拼接到 https://msgchain.org/whitepaper/ 形成完整 URL。
2.4 爬取契约(Crawl Contract)
CRAWL_CONTRACT = {
"primary_seed": "knowledge_network.json",
"module_base_path": "modules/",
"module_filename_field": "entries[*].filename",
"module_url_template": "modules/{filename}",
"module_public_url_template": "https://msgchain.org/whitepaper/modules/{filename}",
"module_export_index": "module_exports/index.json",
"module_export_by_group": "module_exports/by_group.json",
"module_export_by_status": "module_exports/by_status.json",
"module_export_template": "module_exports/{module_stem}.json",
"module_chunk_index": "module_chunks/index.json",
"module_chunk_template": "module_chunks/{module_stem}__chunk_{chunk_no}.json",
"retrieval_hints": "retrieval_hints.json",
"relation_fields": ["outlinks", "backlinks", "related"],
"taxonomy_fields": ["status", "status_label", "group", "group_label", "tags"],
"authoritative_content": (
"Module HTML pages remain the authoritative narrative source; "
"module_exports/*.json provide a machine-friendly derivative layer."
),
}
爬取流程
async def crawl_whitepaper(client: httpx.AsyncClient) -> dict[str, Any]:
# 1. 先读 primary_seed
seed = await client.get(
"https://msgchain.org/whitepaper/knowledge_network.json"
)
seed_data = seed.json()
# 2. 枚举所有模块
modules = {}
for filename, meta in seed_data["entries"].items():
modules[filename] = {
"title": meta["title"],
"status": meta["status"],
"status_label": meta["status_label"],
"group": meta["group"],
"group_label": meta["group_label"],
"tags": meta.get("tags", []),
"outlinks": meta.get("outlinks", []),
"backlinks": meta.get("backlinks", []),
"related": meta.get("related", []),
}
# 3. 读取模块导出索引
export_index = await client.get(
"https://msgchain.org/whitepaper/module_exports/index.json"
)
# 4. 读取分块索引
chunk_index = await client.get(
"https://msgchain.org/whitepaper/module_chunks/index.json"
)
return {
"module_count": seed_data["module_count"],
"modules": modules,
"export_index": export_index.json(),
"chunk_index": chunk_index.json(),
}
knowledge_network.json 是 primary seed,包含所有 65 个模块的元数据、出链、入链和关联关系。
爬取关键规则
- 所有 status 变化(implemented → partial → planned)应触发 AI Agent 重新评估之前基于该模块做出的应答或推理。
- 不要在未读取
agent_entry.json前直接去取modules/*.html——你可能会错过重要的边界信息。 relation_fields(outlinks、backlinks、related)可用于知识图谱推理:当一个模块的信息不足时,跟随 outlinks 跳转到关联模块。
2.5 推荐爬取顺序
RECOMMENDED_CRAWL_ORDER = [
# 层 1:入口和种子
"index.html",
"modules/knowledge_network.html",
"knowledge_network.json",
# 层 2:模块导出
"module_exports/index.json",
"module_exports/by_group.json",
"module_exports/by_status.json",
# 层 3:分块
"module_chunks/index.json",
# 层 4:路由
"retrieval_hints.json",
# 层 5:交付和开发者
"product_delivery_entry.json",
"developer_entry.json",
# 层 6:快速启动
"quickstart/index.json",
"quickstart/contract_and_dapp_minimal.json",
# 层 7:能力矩阵
"developer_capability_matrix.json",
# 层 8:API 规范
"api_specs/index.json",
"api_specs/formal_contracts.json",
"api_specs/rpc_methods.json",
"api_specs/error_codes.json",
# 层 9:OpenAPI
"api_specs/openapi/public_query.yaml",
"api_specs/openapi/contract_surface.yaml",
"api_specs/openapi/agent_surface.yaml",
# 层 10:配方
"recipes/index.json",
"recipes/contract_minimal.json",
"recipes/dapp_minimal.json",
# 层 11:链配置
"chain_config/index.json",
"chain_config/network_presets.json",
"chain_config/developer_sandbox_strategy.json",
# 层 12:合约参考
"contract_reference/index.json",
"contract_reference/core_contracts.json",
# 层 13:模板
"contract_templates/index.json",
# 层 14:示例
"examples/index.json",
# 层 15:发布和执行
"release_pack/index.json",
"e2e_fixtures/index.json",
"execution_pack/index.json",
# 层 16:集成示例
"integration_examples/README.md",
"integration_examples/telegram_bot_crawl_flow.json",
"integration_examples/rag_ingest_flow.json",
"integration_examples/faq_router_prompt_template.md",
"integration_examples/external_ai_agent_bootstrap_prompt.json",
"integration_examples/external_ai_agent_bootstrap_prompt.md",
# 层 17:核心模块 HTML
"modules/executive_summary.html",
"modules/overview.html",
"modules/proof_map.html",
"modules/proof_chain.html",
"modules/trust_verdict.html",
"modules/evidence_index.html",
]
AI Agent 应根据当前任务选择起始层。做问答路由的全部任务从层 4(retrieval_hints.json)开始;做合约开发从层 5(developer_entry.json)开始。
2.6 当前边界
CURRENT_BOUNDARIES = [
"白皮书系统适合机器遍历,但模块结论仍必须服从原始证据边界。",
"implemented 或 partial 只表示当前白皮书口径,不自动等于 MSG 主网 ready。",
"涉及经济、治理、NAT、Explorer、DAO live/public 的结论,必须继续区分本地子门禁与真实生产证据。",
]
这些边界是所有 AI Agent 的硬约束。任何回答都不能违反这三条边界。
2.7 完整入口读取示例
import httpx
import json
from typing import Any
BASE = "https://msgchain.org/whitepaper/"
class WhitepaperAgent:
def __init__(self) -> None:
self.client = httpx.AsyncClient()
self.entry: dict[str, Any] | None = None
self.crawl_contract: dict[str, Any] | None = None
async def bootstrap(self) -> None:
resp = await self.client.get(BASE + "agent_entry.json")
resp.raise_for_status()
self.entry = resp.json()
self.crawl_contract = self.entry["crawl_contract"]
print(f"Discovered schema: {self.entry['schema_version']}")
print(f"Seed: {self.crawl_contract['primary_seed']}")
async def get_knowledge_network(self) -> dict[str, Any]:
resp = await self.client.get(
BASE + self.crawl_contract["primary_seed"]
)
resp.raise_for_status()
return resp.json()
async def get_module_html(self, filename: str) -> str:
url = self.crawl_contract["module_public_url_template"].format(
filename=filename
)
resp = await self.client.get(url)
resp.raise_for_status()
return resp.text
async def get_module_export(self, stem: str) -> dict[str, Any]:
url = self.crawl_contract["module_export_public_url_template"].format(
module_stem=stem
)
resp = await self.client.get(url)
resp.raise_for_status()
return resp.json()
async def get_chunk(self, stem: str, chunk_no: int) -> dict[str, Any]:
url = self.crawl_contract["module_chunk_public_url_template"].format(
module_stem=stem, chunk_no=f"{chunk_no:02d}"
)
resp = await self.client.get(url)
resp.raise_for_status()
return resp.json()
async def close(self) -> None:
await self.client.aclose()
3. 模块导出与分块
白皮书内容分为三个层次,AI Agent 应根据检索需求选择适当层。
3.1 三层内容层级
from enum import Enum
class ContentLayer(Enum):
MODULE_HTML = 1
MODULE_EXPORT = 2
MODULE_CHUNK = 3
CONTENT_LAYERS = {
ContentLayer.MODULE_HTML: {
"name": "模块 HTML",
"path": "modules/*.html",
"description": "权威叙事源——完整的人类可读白皮书页面,包含流程图、铁证和原始证据引用。",
"best_for": "深度理解、完整上下文查询、证据验证",
"language": "中文+英文术语",
},
ContentLayer.MODULE_EXPORT: {
"name": "模块导出 JSON",
"path": "module_exports/*.json",
"description": "机器友好的衍生层——从 HTML 提取的结构化元数据、键值对、边界子句和引用。",
"best_for": "元数据查询、标签分组、状态过滤、关联分析",
"language": "JSON,纯机器格式",
},
ContentLayer.MODULE_CHUNK: {
"name": "模块分块",
"path": "module_chunks/*.json",
"description": "检索友好的分段——将 HTML 切成 2 个分块,每个分块包含文本块、标签和边界子句。",
"best_for": "RAG 管道、分块式问答、快速上下文检索",
"language": "中文文本块,JSON 包装",
},
}
何时使用哪一层
def recommend_layer(query_type: str) -> ContentLayer:
mapping = {
"factoid_qa": ContentLayer.MODULE_CHUNK,
"topic_routing": ContentLayer.MODULE_EXPORT,
"deep_understanding": ContentLayer.MODULE_HTML,
"evidence_collection": ContentLayer.MODULE_HTML,
"status_check": ContentLayer.MODULE_EXPORT,
"relationship_graph": ContentLayer.MODULE_EXPORT,
"rag_ingestion": ContentLayer.MODULE_CHUNK,
"code_generation": ContentLayer.MODULE_EXPORT,
}
return mapping.get(query_type, ContentLayer.MODULE_CHUNK)
3.2 模块导出索引
module_exports/index.json 列出了所有模块的张成。每个导出包含:
async def load_module_exports(client: httpx.AsyncClient) -> dict[str, Any]:
resp = await client.get(
"https://msgchain.org/whitepaper/module_exports/index.json"
)
resp.raise_for_status()
return resp.json()
# 返回结构: { "modules": [...], "count": 65 }
# 每条记录包含:
# { "filename", "title", "status", "status_label", "group", "group_label", "tags" }
AI Agent 可以通过 module_exports/by_group.json 按 group 过滤,通过 module_exports/by_status.json 按 status 过滤。
async def filter_modules_by_status(
client: httpx.AsyncClient, target_status: str
) -> list[dict[str, Any]]:
resp = await client.get(
f"https://msgchain.org/whitepaper/module_exports/by_status.json"
)
data = resp.json()
return data.get(target_status, [])
# 可能的 status: "implemented", "partial", "planned", "audit"
3.3 模块分块索引
module_chunks/index.json 是所有分块的索引。每个分块文件命名约定为:
module_chunks/{module_stem}__chunk_{chunk_no}.json
例如 module_chunks/contract__chunk_01.json、module_chunks/contract__chunk_02.json。
async def load_chunk_index(client: httpx.AsyncClient) -> dict[str, Any]:
resp = await client.get(
"https://msgchain.org/whitepaper/module_chunks/index.json"
)
resp.raise_for_status()
return resp.json()
# 返回结构: { "chunks": [{"chunk_id": "...", "module": "...", "chunk_no": 1, ...}], "count": 130 }
3.4 分块检索策略
from dataclasses import dataclass
@dataclass
class ChunkInfo:
chunk_id: str
module_title: str
module_status: str
excerpt: str
public_url: str
async def retrieve_chunks_for_question(
client: httpx.AsyncClient, topic_id: str, retrieval_hints: dict
) -> list[ChunkInfo]:
"""根据 retrieval_hints 中的 recommended_chunks 获取分块。"""
topics = {t["topic_id"]: t for t in retrieval_hints["topics"]}
topic = topics.get(topic_id)
if not topic:
return []
chunks = []
for ref in topic.get("recommended_chunks", [])[:4]:
resp = await client.get(ref["chunk_public_url"])
if resp.status_code != 200:
continue
chunk_data = resp.json()
chunks.append(ChunkInfo(
chunk_id=ref["chunk_id"],
module_title=chunk_data.get("module_title", ""),
module_status=chunk_data.get("module_status", ""),
excerpt=ref.get("excerpt", ""),
public_url=ref["chunk_public_url"],
))
return chunks
# 返回的分块已按 implement > partial > planned 的优先级排序
分块加载到 RAG 向量库
async def ingest_all_chunks_to_vector_store(client: httpx.AsyncClient) -> list[dict]:
"""批量加载所有分块,准备向量化。"""
chunk_index = await load_chunk_index(client)
documents = []
for chunk_ref in chunk_index["chunks"]:
resp = await client.get(chunk_ref["chunk_public_url"])
if resp.status_code != 200:
continue
chunk = resp.json()
documents.append({
"id": chunk_ref["chunk_id"],
"text": chunk.get("text", ""),
"module_title": chunk.get("module_title", ""),
"module_filename": chunk.get("module_filename", ""),
"module_status": chunk.get("module_status", ""),
"module_status_label": chunk.get("module_status_label", ""),
"module_group": chunk.get("module_group", ""),
"module_group_label": chunk.get("module_group_label", ""),
"module_tags": chunk.get("module_tags", []),
"boundary_clauses": chunk.get("boundary_clauses", []),
"evidence_refs": chunk.get("evidence_refs", []),
"module_public_url": chunk.get("module_public_url", ""),
"chunk_public_url": chunk_ref["chunk_public_url"],
})
return documents
建议在向量化时,将 module_status、module_tags 和 boundary_clauses 作为元数据字段存入向量库,以便在检索时做过滤和边界检查。
4. 检索提示:retrieval_hints.json
retrieval_hints.json 是 AI Agent 的主题路由引擎。它将常见问题映射到白皮书中的相关模块和分块。
4.1 主题概览
RETRIEVAL_TOPICS = [
{
"topic_id": "economics",
"title": "经济模型与发行结算",
"questions": [
"MSG 的发行规则是什么",
"economic seconds 如何结算",
"候选池和验证者池如何分配",
],
"preferred_tags": ["铸造", "奖励", "结算"],
},
{
"topic_id": "treasury_governance",
"title": "基金会金库与治理执行",
"questions": [
"基金会金库怎么执行",
"DAO 如何控制 treasury",
"timelock 和阈值怎么配合",
],
"preferred_tags": ["金库", "多签", "治理", "时间锁"],
},
{
"topic_id": "consensus_validator",
"title": "共识、DAR 与验证者资格",
"questions": [
"MSG 如何轮值出块",
"DAR 怎么影响验证者",
"处罚和恢复机制是什么",
],
"preferred_tags": ["处罚", "挑战", "恢复"],
},
{
"topic_id": "ai_agent_runtime",
"title": "AI Agent、控制面与任务闭环",
"questions": [
"MSG 的 AI Agent 能做什么",
"AI Task L2 如何落链",
"AI 控制平面和治理如何隔离",
],
"preferred_tags": ["AI", "Agent"],
},
{
"topic_id": "explorer_data_access",
"title": "Explorer、查询面与数据检索",
"questions": [
"MSG 有哪些 Explorer 能力",
"如何查 contract source",
"eth_getLogs 和 WS logs 支持到什么程度",
],
"preferred_tags": ["接口", "查询", "索引"],
},
{
"topic_id": "contract_development",
"title": "智能合约开发与部署闭环",
"questions": [
"如何在 MSG 上开发 CosmWasm 合约",
"怎样部署和调用智能合约",
"StoreCode / Instantiate / Execute / Query 怎么走",
],
"preferred_tags": ["SDK", "开发者", "接口", "查询"],
},
{
"topic_id": "dapp_integration",
"title": "dApp 前端接入与钱包集成",
"questions": [
"如何开发 MSG dApp",
"Keplr / CosmJS 怎么接入",
"前端怎样查询、签名和回看 receipt",
],
"preferred_tags": ["SDK", "开发者", "接口", "查询"],
},
{
"topic_id": "security_quantum",
"title": "安全、抗量子与基础设施",
"questions": [
"MSG 用什么抗量子算法",
"节点安全怎么做",
"P2P 和 NAT 穿透如何工作",
],
"preferred_tags": ["安全", "抗量子", "P2P", "组网", "NAT"],
},
]
4.2 问题路由实现
from rapidfuzz import process as fuzzy_process
from typing import Any
class QuestionRouter:
def __init__(self, retrieval_hints: dict[str, Any]):
self.hints = retrieval_hints
self.topics: list[dict] = retrieval_hints["topics"]
self._build_index()
def _build_index(self) -> None:
self.question_map: list[tuple[str, str, str]] = []
for topic in self.topics:
for q in topic.get("questions", []):
self.question_map.append((q, topic["topic_id"], topic["title"]))
def route(self, question: str, threshold: int = 60) -> list[dict[str, Any]]:
"""将用户问题匹配到最相关的主题。"""
if not self.question_map:
return []
results = fuzzy_process.extract(
question,
[q for q, _, _ in self.question_map],
score_cutoff=threshold,
)
matched_topics = {}
for match_text, score, _ in results:
idx = next(i for i, (q, _, _) in enumerate(self.question_map) if q == match_text)
_, topic_id, title = self.question_map[idx]
if topic_id not in matched_topics or score > matched_topics[topic_id]["score"]:
matched_topics[topic_id] = {"topic_id": topic_id, "title": title, "score": score}
return sorted(matched_topics.values(), key=lambda x: x["score"], reverse=True)
def get_recommended_modules(self, topic_id: str) -> list[dict]:
for topic in self.topics:
if topic["topic_id"] == topic_id:
return topic.get("recommended_modules", [])
return []
def get_recommended_chunks(self, topic_id: str) -> list[dict]:
for topic in self.topics:
if topic["topic_id"] == topic_id:
return topic.get("recommended_chunks", [])
return []
4.3 带边界的答案生成
每个主题都有固定的 answering_guidance,AI Agent 必须遵守:
ANSWERING_GUIDANCE = (
"先回答当前已实现或已证实的部分,再明确未完成边界。"
"若涉及经济、治理、Explorer 或 live/public 结论,优先引用带 evidence_refs 与 boundary_clauses 的模块或 chunk。"
"不要把 partial、planned 或本地子门禁关闭偷换成主网 ready。"
)
async def answer_with_hints(
router: QuestionRouter,
client: httpx.AsyncClient,
question: str,
) -> dict[str, Any]:
"""完整的检索-路由-答案生成流程。"""
matches = router.route(question)
if not matches:
return {
"answer": "当前白皮书机器层没有匹配的主题。",
"confidence": "low",
"boundaries": ["该问题超出 retrieval_hints.json 覆盖范围。"],
}
best = matches[0]
recommended_chunks = router.get_recommended_chunks(best["topic_id"])
recommended_modules = router.get_recommended_modules(best["topic_id"])
chunk_contexts = []
for ref in recommended_chunks[:3]:
resp = await client.get(ref["chunk_public_url"])
if resp.status_code == 200:
chunk = resp.json()
chunk_contexts.append({
"chunk_id": ref["chunk_id"],
"text": chunk.get("text", "")[:500],
"boundary_clauses": chunk.get("boundary_clauses", []),
})
return {
"matched_topic": best["title"],
"topic_id": best["topic_id"],
"confidence": "high" if len(chunk_contexts) > 0 else "medium",
"chunks_loaded": len(chunk_contexts),
"boundaries": [
clause for c in chunk_contexts
for clause in c.get("boundary_clauses", [])
],
"answering_guidance": ANSWERING_GUIDANCE,
}
4.4 推荐模块与分块的读取顺序
当一个主题有多个推荐模块时,按 status 优先级读取:
implemented > partial > planned > audit
MODULE_STATUS_PRIORITY = {
"implemented": 0,
"partial": 1,
"planned": 2,
"audit": 3,
}
async def fetch_best_modules(
client: httpx.AsyncClient, modules: list[dict]
) -> list[dict]:
"""按 status 排序后获取模块导出内容。"""
sorted_modules = sorted(
modules,
key=lambda m: MODULE_STATUS_PRIORITY.get(m.get("status", "planned"), 99),
)
results = []
for mod in sorted_modules[:4]:
stem = mod["filename"].replace(".html", "")
url = f"https://msgchain.org/whitepaper/module_exports/{stem}.json"
resp = await client.get(url)
if resp.status_code == 200:
results.append(resp.json())
return results
5. 开发者入口:developer_entry.json
developer_entry.json 是 AI Agent 进行 MSG Chain 合约和 dApp 开发的核心引导文件。它定义了三种引导顺序,覆盖不同的开发场景。
5.1 关键模块
KEY_DEVELOPER_MODULES = [
{"filename": "contract.html", "title": "智能合约引擎 (CosmWasm)", "status": "implemented", "group": "control"},
{"filename": "rpc.html", "title": "RPC 与 API 接口", "status": "implemented", "group": "runtime"},
{"filename": "registry.html", "title": "创世注册中心 (genesis_registry_v1)", "status": "implemented", "group": "control"},
{"filename": "keplr.html", "title": "生态层: Web3 钱包集成", "status": "partial", "group": "ecosystem"},
{"filename": "explorer.html", "title": "生态层: 区块浏览器与数据索引", "status": "partial", "group": "ecosystem"},
{"filename": "agent_api_surface.html", "title": "Agent API 能力表面", "status": "partial", "group": "developer"},
{"filename": "ai_wallet.html", "title": "AI 智能钱包闭环", "status": "partial", "group": "ai"},
{"filename": "sdk_dev_surface.html", "title": "SDK 与开发者能力表面", "status": "partial", "group": "developer"},
]
5.2 推荐合约引导顺序(19 步)
CONTRACT_BOOTSTRAP_ORDER = [
"product_delivery_entry.json",
"developer_capability_matrix.json",
"quickstart/contract_and_dapp_minimal.json",
"chain_config/index.json",
"chain_config/developer_sandbox_strategy.json",
"api_specs/rpc_methods.json",
"api_specs/openapi/contract_surface.yaml",
"api_specs/formal_contracts.json",
"contract_reference/index.json",
"contract_reference/core_contracts.json",
"contract_templates/index.json",
"recipes/contract_minimal.json",
"execution_pack/index.json",
"execution_pack/command_registry.json",
"e2e_fixtures/index.json",
"modules/contract.html",
"modules/registry.html",
"modules/rpc.html",
"module_exports/contract.json",
]
async def bootstrap_contract_development(
client: httpx.AsyncClient,
) -> dict[str, Any]:
"""按 recommended_contract_bootstrap_order 加载开发资产。"""
base = "https://msgchain.org/whitepaper/"
loaded = {}
for path in CONTRACT_BOOTSTRAP_ORDER:
url = base + path
resp = await client.get(url)
if resp.status_code == 200:
loaded[path] = resp.json()
else:
print(f"Warning: {url} returned {resp.status_code}")
return loaded
各步骤目的
CONTRACT_STEP_PURPOSE = {
"product_delivery_entry.json": "了解产品交付的全生命周期和边界",
"developer_capability_matrix.json": "评估各开发表面的机器就绪度",
"quickstart/contract_and_dapp_minimal.json": "获取 7 步快速启动的最小路径",
"chain_config/index.json": "读取链配置清单",
"chain_config/developer_sandbox_strategy.json": "确认 sandbox 策略——默认本地 fail-closed",
"api_specs/rpc_methods.json": "获取 RPC 方法列表与签名",
"api_specs/openapi/contract_surface.yaml": "获取合约面 OpenAPI 摘要",
"api_specs/formal_contracts.json": "获取正式 API / Schema 契约索引",
"contract_reference/index.json": "获取合约参考索引",
"contract_reference/core_contracts.json": "获取核心合约消费索引与 interface/source pack",
"contract_templates/index.json": "获取合约模板索引",
"recipes/contract_minimal.json": "获取最小合约配方",
"execution_pack/index.json": "获取执行包入口",
"execution_pack/command_registry.json": "获取命令注册表",
"e2e_fixtures/index.json": "获取端到端测试夹具",
"modules/contract.html": "读取合约引擎权威叙事",
"modules/registry.html": "读取注册中心叙事",
"modules/rpc.html": "读取 RPC 叙事",
"module_exports/contract.json": "读取合约模块的结构化导出",
}
5.3 推荐 dApp 引导顺序(17 步)
DAPP_BOOTSTRAP_ORDER = [
"product_delivery_entry.json",
"developer_capability_matrix.json",
"quickstart/contract_and_dapp_minimal.json",
"chain_config/index.json",
"chain_config/developer_sandbox_strategy.json",
"api_specs/rpc_methods.json",
"api_specs/openapi/public_query.yaml",
"api_specs/formal_contracts.json",
"examples/index.json",
"recipes/dapp_minimal.json",
"execution_pack/index.json",
"execution_pack/command_registry.json",
"release_pack/index.json",
"modules/keplr.html",
"modules/rpc.html",
"modules/explorer.html",
"module_exports/keplr.json",
]
async def bootstrap_dapp_development(
client: httpx.AsyncClient,
) -> dict[str, Any]:
"""按 recommended_dapp_bootstrap_order 加载 dApp 开发资产。"""
base = "https://msgchain.org/whitepaper/"
loaded = {}
for path in DAPP_BOOTSTRAP_ORDER:
url = base + path
resp = await client.get(url)
if resp.status_code == 200:
loaded[path] = resp.json()
return loaded
5.4 推荐全生命周期顺序(19 步)
FULL_LIFECYCLE_ORDER = [
"product_delivery_entry.json",
"developer_entry.json",
"quickstart/index.json",
"quickstart/contract_and_dapp_minimal.json",
"chain_config/developer_sandbox_strategy.json",
"contract_reference/index.json",
"contract_reference/core_contracts.json",
"api_specs/formal_contracts.json",
"integration_examples/external_ai_agent_bootstrap_prompt.json",
"developer_capability_matrix.json",
"chain_config/index.json",
"contract_templates/index.json",
"examples/index.json",
"execution_pack/index.json",
"execution_pack/delivery_workflows.json",
"execution_pack/command_registry.json",
"release_pack/index.json",
"e2e_fixtures/index.json",
"modules/review_playbook.html",
"modules/evidence_index.html",
]
全生命周期适用于从零到交付的场景。它涵盖 scope → codegen → quality_gate → deploy_and_verify → launch 五个阶段。
5.5 必须的人类输入
HUMAN_INPUTS_REQUIRED = [
"产品目标与业务规则",
"真实部署权限与签名账户",
"生产环境变量、域名、CI/CD 或发布权限",
"治理、多签、金库、审批等高风险动作的授权与窗口",
]
AI Agent 不能在这些方面做自动决策。所有涉及上述内容的操作都必须停顿并请求人类确认。
5.6 当前开发者边界
DEVELOPER_BOUNDARIES = [
"当前开发协议层可显著提升 AI coding 的可执行性,但仍不能诚实承诺只靠入口即可 100% 自动完成任何产品上线。",
"当前已补 Quick Start、source-backed 合约消费索引、正式 API/Schema 契约索引与 fail-closed sandbox 策略,但仍不等于 signed public SDK、public sandbox 或 not independently verified for production 交付。",
"涉及私钥、部署权限、生产域名、资金操作、DAO/timelock/threshold 的动作,必须保留人类确认与审批门禁。",
]
5.7 完整开发者引导实现
from enum import Enum
class BootstrapMode(Enum):
CONTRACT = "contract"
DAPP = "dapp"
FULL_LIFECYCLE = "full_lifecycle"
BOOTSTRAP_ORDERS = {
BootstrapMode.CONTRACT: CONTRACT_BOOTSTRAP_ORDER,
BootstrapMode.DAPP: DAPP_BOOTSTRAP_ORDER,
BootstrapMode.FULL_LIFECYCLE: FULL_LIFECYCLE_ORDER,
}
async def guided_developer_bootstrap(
client: httpx.AsyncClient,
mode: BootstrapMode,
) -> dict[str, Any]:
"""按模式加载对应的引导顺序。"""
base = "https://msgchain.org/whitepaper/"
order = BOOTSTRAP_ORDERS[mode]
result = {"mode": mode.value, "steps_completed": 0, "assets": {}}
for i, path in enumerate(order, 1):
url = base + path
resp = await client.get(url)
if resp.status_code == 200:
content_type = resp.headers.get("content-type", "")
if "json" in content_type:
result["assets"][path] = resp.json()
elif "yaml" in content_type or "yaml" in path:
result["assets"][path] = resp.text
elif "md" in path:
result["assets"][path] = resp.text
result["steps_completed"] += 1
print(f"[{i}/{len(order)}] Loaded {path}")
else:
print(f"[{i}/{len(order)}] FAILED {path} ({resp.status_code})")
return result
6. 能力矩阵:developer_capability_matrix.json
developer_capability_matrix.json 定义了 16 个开发者能力表面,每个表面都有 machine_readiness 等级,告诉 AI Agent 哪些可以安全自动处理、哪些需要人工介入。
6.1 16 个能力表面
SURFACE_COUNT = 16
SURFACE_IDS = [
"contract_runtime",
"core_contract_reference_pack",
"registry_resolution",
"rpc_gateway",
"formal_api_schema_pack",
"wallet_frontend",
"explorer_receipts",
"agent_query_and_guarded_write",
"sdk_surface",
"chain_config_pack",
"public_sandbox_strategy",
"contract_template_pack",
"dapp_starter_pack",
"e2e_test_fixture_pack",
"execution_workflow_pack",
"release_delivery_pack",
]
6.2 机器就绪度(machine_readiness)等级
MACHINE_READINESS_LEVELS = {
"assisted_codegen": {
"rank": 1,
"meaning": "AI 可辅助代码生成,write_path_ready = true",
"safe_for_automation": True,
},
"source_backed_reference": {
"rank": 2,
"meaning": "基于源码的参考索引,可读不可写",
"safe_for_automation": True,
},
"production_reference": {
"rank": 3,
"meaning": "开发参考级别的参考数据,只读",
"safe_for_automation": True,
},
"starter_ready": {
"rank": 4,
"meaning": "起始模板就绪,可生成骨架代码",
"safe_for_automation": True,
},
"guarded_integration": {
"rank": 5,
"meaning": "受保护的集成路径,需人工确认",
"safe_for_automation": False,
},
"guarded_write": {
"rank": 6,
"meaning": "受保护的写路径,需签批",
"safe_for_automation": False,
},
"read_only_assist": {
"rank": 7,
"meaning": "只读辅助,不能自动触发写操作",
"safe_for_automation": False,
},
"fail_closed_reference": {
"rank": 8,
"meaning": "默认关闭的参考策略,需显式开启",
"safe_for_automation": False,
},
"local_candidate": {
"rank": 9,
"meaning": "本地候选,不可用于生产",
"safe_for_automation": False,
},
}
6.3 各表面详情
SURFACE_DETAILS = {
"contract_runtime": {
"title": "CosmWasm 合约运行时与生命周期",
"machine_readiness": "assisted_codegen",
"write_path_ready": True,
"schema_available": True,
"example_available": True,
"production_supported": True,
"best_for": ["合约消息设计", "状态流转设计", "部署调用路径生成", "回执验证规划"],
"blocking_gaps": [
"当前模板是 starter pack,不等于官方业务合约全集",
"缺少从链实现自动导出的正式 schema 契约",
"生产部署仍需真实权限与验收",
],
"boundaries": [
"运行时主线已实现,且已补 starter template,但不等于所有业务模板都已完善。",
"AI 可辅助生成合约代码,但仍需结合业务规则、人类审核与真实部署账户完成落地。",
],
},
"core_contract_reference_pack": {
"title": "核心合约消费索引与 interface/source pack",
"machine_readiness": "source_backed_reference",
"write_path_ready": False,
"schema_available": True,
"example_available": True,
"production_supported": False,
"best_for": [
"识别核心合约 canonical key 与 schema",
"生成 typed query/execute payload",
"定位 msg.rs / schema / Cargo 元数据",
],
"blocking_gaps": [
"当前公开的是 interface/source 消费索引,不是全部核心合约逻辑公开包",
"signed schema release 与 canonical mapping cross-check 仍未关闭",
],
"boundaries": [
"当前适合外部 AI 读取核心合约接口、schema 与 message source,但不能误当成核心合约全量公开与 not independently verified for production ABI。",
"store/instantiate/migrate 等高风险动作仍必须服从治理、审批与真实 receipt 证据。",
],
},
"registry_resolution": {
"title": "Registry Canonical Key 与地址解析",
"machine_readiness": "production_reference",
"write_path_ready": False,
"schema_available": False,
"example_available": True,
"production_supported": True,
"best_for": ["合约地址寻址", "canonical key 解析", "部署后地址发现"],
"boundaries": ["可作为机器寻址入口,但仍应以真实 query 返回为准。"],
},
"rpc_gateway": {
"title": "RPC / REST / gRPC 查询与广播网关",
"machine_readiness": "assisted_codegen",
"write_path_ready": True,
"schema_available": True,
"example_available": True,
"production_supported": True,
"best_for": ["dApp 查询层", "交易广播", "基础客户端封装"],
"blocking_gaps": [
"当前 OpenAPI 仍是文档流水线维护的协议摘要",
"live/public 端点仍需以真实可用性与证据为准",
],
"boundaries": ["当前已补机器可消费 OpenAPI 摘要,但仍不是从生产服务自动导出的最终 API 契约。"],
},
"wallet_frontend": {
"title": "Keplr / CosmJS / 钱包前端接入",
"machine_readiness": "guarded_integration",
"write_path_ready": True,
"schema_available": False,
"example_available": True,
"production_supported": False,
"best_for": ["dApp 钱包接入", "chain config 注入", "签名与查询前端路径"],
"blocking_gaps": ["缺少对外稳定前端样例项目", "兼容目标不等于生态全量上线"],
"boundaries": ["当前是兼容路径与配置底座,不应表述成官方钱包生态已全量上线。"],
},
"explorer_receipts": {
"title": "Explorer 查询、receipt 与日志检索面",
"machine_readiness": "read_only_assist",
"write_path_ready": False,
"schema_available": True,
"example_available": True,
"production_supported": False,
"best_for": ["receipt 回放", "事件追踪", "contract source 元数据定位"],
"blocking_gaps": [
"live/public Explorer 仍未全量关闭",
"索引与分页的生产 SLO/保留策略仍需独立交付",
],
"boundaries": ["可辅助开发调试与回证,不应直接当成完全生产化 Explorer 平台。"],
},
"agent_query_and_guarded_write": {
"title": "Agent API 查询面与受保护写路径",
"machine_readiness": "guarded_write",
"write_path_ready": False,
"schema_available": False,
"example_available": False,
"production_supported": False,
"best_for": ["AI coding 辅助编排", "监控读取", "高风险动作门禁理解"],
"blocking_gaps": [
"部分写路径仍为 stub",
"缺少稳定 OpenAPI",
"自动执行仍需审批、风控、密钥与权限系统",
],
"boundaries": [
"不能把 Agent 写路径当成已完成的全自动生产执行面。",
"所有高风险写动作仍需区分真实写路径、受保护路径与 stub 路径。",
],
},
"sdk_surface": {
"title": "MSG SDK 开发表面",
"machine_readiness": "local_candidate",
"write_path_ready": False,
"schema_available": False,
"example_available": False,
"production_supported": False,
"best_for": ["识别 alpha SDK 分层", "为后续签名发布留出能力映射"],
"blocking_gaps": ["尚未形成 signed/public SDK release", "白皮书机器层仍不直接承载 SDK 安装证明"],
"boundaries": ["SDK 已进入 alpha/local candidate,但不能被 AI 误当成稳定公共客户端。"],
},
}
6.4 自动化的安全范围判断
def can_automate(surface: dict) -> bool:
"""判断 AI 是否可以安全地自动化这个表面的操作。"""
readiness = surface.get("machine_readiness", "")
allowed = {
"assisted_codegen",
"source_backed_reference",
"production_reference",
"starter_ready",
}
return readiness in allowed
def needs_human_approval(surface: dict) -> bool:
"""判断这个表面的写操作是否需要人类审批。"""
return (
not surface.get("write_path_ready", False)
or surface.get("machine_readiness") in ("guarded_write", "guarded_integration")
)
async def generate_safety_report(
client: httpx.AsyncClient,
) -> list[dict[str, Any]]:
"""加载能力矩阵并生成自动化安全报告。"""
resp = await client.get(
"https://msgchain.org/whitepaper/developer_capability_matrix.json"
)
matrix = resp.json()
report = []
for item in matrix["items"]:
report.append({
"surface_id": item["surface_id"],
"title": item["title"],
"machine_readiness": item["machine_readiness"],
"can_automate": can_automate(item),
"needs_human_approval": needs_human_approval(item),
"blocking_gaps": item.get("blocking_gaps", []),
"boundaries": item.get("boundaries", []),
})
return report
6.5 能力矩阵的消费策略
AI Agent 在决定某个开发操作的自动化程度时,应:
- 先查能力矩阵,找到对应
surface_id。 - 检查
machine_readiness等级——如果不在 safe_for_automation 列表里,标记为需人工介入。 - 即使
machine_safe_for_codegen = true,也要检查write_path_ready——如果为 false,生成的代码不能自动部署。 - 引用
boundaries和blocking_gaps在回答中,不隐藏已知限制。
7. 快速启动:quickstart/
quickstart/ 目录为 AI Agent 提供了最小化的快速上手路径。
7.1 contract_and_dapp_minimal.json(7 步)
QUICKSTART_STEPS = [
{
"step": 1,
"action": "read capability + sandbox boundaries",
"consumes": [
"developer_entry.json",
"chain_config/developer_sandbox_strategy.json",
],
"purpose": "先判断 public sandbox 是否可用,默认按 local fail-closed 路径起步。",
},
{
"step": 2,
"action": "read contract interface pack",
"consumes": [
"contract_reference/core_contracts.json",
"api_specs/formal_contracts.json",
],
"purpose": "先拿到核心合约 schema、msg.rs、OpenAPI 与 tool manifest。",
},
{
"step": 3,
"action": "scaffold contract",
"consumes": [
"contract_templates/index.json",
"recipes/contract_minimal.json",
],
"purpose": "从 starter template 进入,再按核心合约 interface pack 补业务消息模型。",
},
{
"step": 4,
"action": "scaffold dapp",
"consumes": [
"examples/index.json",
"recipes/dapp_minimal.json",
"chain_config/network_presets.json",
],
"purpose": "生成钱包、query、execute adapter 与最小前端结构。",
},
{
"step": 5,
"action": "bind source-backed API and schema contracts",
"consumes": ["api_specs/formal_contracts.json"],
"purpose": "把 request/response、receipt/event、contract schema 与 error code 统一到正式索引。",
},
{
"step": 6,
"action": "follow local starter and E2E boundary",
"consumes": [
"quickstart/ai_agent_dapp_starter_README.md",
"release_pack/developer_dapp_starter_e2e_manifest.json",
],
"purpose": "先走 local verified path,不把 local starter 误当 public E2E。",
},
{
"step": 7,
"action": "prepare guarded release plan",
"consumes": [
"release_pack/index.json",
"execution_pack/index.json",
"e2e_fixtures/index.json",
],
"purpose": "形成命令、验收、receipt/query/log 证据计划。",
},
]
7.2 命令入口点
COMMAND_ENTRY_POINTS = {
"deps": {
"path": "execution_pack/command_registry.json",
"when_to_use": "初始化依赖与本地开发环境",
},
"lint": {
"path": "execution_pack/command_registry.json",
"when_to_use": "进入质量门禁前先做静态检查",
},
"test": {
"path": "execution_pack/command_registry.json",
"when_to_use": "验证 Go 主体功能",
},
"ci_contracts": {
"path": "execution_pack/command_registry.json",
"when_to_use": "统一跑合约构建与 cargo test",
},
"build_linux": {
"path": "execution_pack/command_registry.json",
"when_to_use": "需要生成节点二进制或部署制品时",
},
}
7.3 工作流入口点
WORKFLOW_ENTRY_POINTS = {
"contract_delivery_guarded_v1": {
"path": "execution_pack/delivery_workflows.json",
"description": "合约交付受保护工作流",
},
"dapp_delivery_guarded_v1": {
"path": "execution_pack/delivery_workflows.json",
"description": "dApp 交付受保护工作流",
},
"product_launch_guarded_v1": {
"path": "execution_pack/delivery_workflows.json",
"description": "产品发布受保护工作流",
},
}
7.4 最小本地命令序列
MINIMAL_LOCAL_COMMAND_SEQUENCE = [
{"order": 1, "command_id": "deps", "why": "准备依赖环境"},
{"order": 2, "command_id": "lint", "why": "尽早发现静态错误"},
{"order": 3, "command_id": "test", "why": "验证主仓库逻辑"},
{"order": 4, "command_id": "ci_contracts", "why": "验证合约 build 与测试闭环"},
]
async def execute_minimal_local_sequence(client: httpx.AsyncClient) -> list[dict]:
"""加载命令注册表并执行最小本地序列。"""
resp = await client.get(
"https://msgchain.org/whitepaper/execution_pack/command_registry.json"
)
registry = resp.json()
results = []
for step in MINIMAL_LOCAL_COMMAND_SEQUENCE:
cmd_id = step["command_id"]
if cmd_id in registry:
cmd_spec = registry[cmd_id]
results.append({
"step": step["order"],
"command_id": cmd_id,
"command": cmd_spec.get("command", ""),
"purpose": step["why"],
"loaded_spec": True,
})
else:
results.append({
"step": step["order"],
"command_id": cmd_id,
"error": f"Command {cmd_id} not found in registry",
})
return results
7.5 本地启动资产
LOCAL_STARTER_ASSETS = {
"ai_agent_dapp_starter_readme": {
"path": "quickstart/ai_agent_dapp_starter_README.md",
"public_url": "https://msgchain.org/whitepaper/quickstart/ai_agent_dapp_starter_README.md",
},
"ai_agent_dapp_starter_env": {
"path": "quickstart/ai_agent_dapp_starter.env.example",
"public_url": "https://msgchain.org/whitepaper/quickstart/ai_agent_dapp_starter.env.example",
},
}
7.6 Quick Start 边界
QUICKSTART_BOUNDARIES = [
"Quick Start 的目标是让外部 AI 更快进入正确入口,不是替代正式生产契约、公共测试网或最终上线审批。",
"当前 public sandbox 默认为 disabled/fail-closed;没有显式 public proof 前,AI 必须使用 local-only 路径并保留 No-Go 边界。",
]
8. 集成示例
8.1 Telegram 机器人爬取流程
适用于第三方 Telegram 投资群或技术群问答机器人。
TELEGRAM_CRAWL_FLOW = {
"entry_points": [
"../agent_entry.json",
"../retrieval_hints.json",
"../module_chunks/index.json",
],
"recommended_flow": [
{
"step": 1,
"action": "load_agent_entry",
"purpose": "发现当前稳定的机器入口、URL 映射和爬取顺序。",
},
{
"step": 2,
"action": "route_question",
"purpose": "将用户问题与 retrieval_hints.json 中的主题问题和偏好标签匹配。",
},
{
"step": 3,
"action": "fetch_recommended_chunks",
"purpose": "读取前 2-4 个推荐的分块作为快速答案上下文。",
},
{
"step": 4,
"action": "fetch_recommended_modules_if_needed",
"purpose": "当答案需要更广的上下文或更强的证实时,获取模块导出或 HTML。",
},
{
"step": 5,
"action": "answer_with_boundaries",
"purpose": "返回带有明确实现边界的答案,避免高估就绪度。",
},
],
"reply_policy": [
"优先使用 implemented 模块而非 partial 或 planned",
"如果只有 partial 或 planned 材料,明确说明",
"对经济、治理、Explorer 和 live/public 声明,至少包含一条边界子句",
"如果问题涉及操作敏感内容,引导用户阅读权威模块 HTML 页面",
],
}
class TelegramBotCrawler:
def __init__(self) -> None:
self.client = httpx.AsyncClient()
self.router: QuestionRouter | None = None
async def bootstrap(self) -> None:
hints_resp = await self.client.get(
"https://msgchain.org/whitepaper/retrieval_hints.json"
)
hints = hints_resp.json()
self.router = QuestionRouter(hints)
async def handle_question(self, question: str) -> str:
assert self.router is not None
matches = self.router.route(question)
if not matches:
return "抱歉,当前白皮书机器层没有匹配到此问题的主题。"
best = matches[0]
chunks = self.router.get_recommended_chunks(best["topic_id"])
context = []
for ref in chunks[:2]:
resp = await self.client.get(ref["chunk_public_url"])
if resp.status_code == 200:
context.append(resp.json().get("text", "")[:300])
boundaries = [
"注意:这是基于白皮书机器层的回答,partial/planned 模块不视为生产已就绪。"
]
answer_parts = [f"主题:{best['title']}"]
if context:
answer_parts.append("\n".join(context[:2]))
answer_parts.append("\n".join(boundaries))
return "\n\n".join(answer_parts)
async def close(self) -> None:
await self.client.aclose()
8.2 RAG 摄取流程
适用于向量数据库、检索服务或离线索引管线。
RAG_INGEST_FLOW = {
"batch_flow": [
{
"step": 1,
"action": "load_manifest",
"input": "../whitepaper_manifest.json",
"purpose": "捕获当前 schema 版本、生成时间和入口点。",
},
{
"step": 2,
"action": "load_module_index",
"input": "../module_exports/index.json",
"purpose": "枚举所有模块并保留 status/group 元数据。",
},
{
"step": 3,
"action": "load_chunk_index",
"input": "../module_chunks/index.json",
"purpose": "枚举所有分块文件并映射回所属模块。",
},
{
"step": 4,
"action": "ingest_chunk_documents",
"input": "../module_chunks/*.json",
"purpose": "使用分块文本作为向量索引的主要可检索文档主体。",
},
{
"step": 5,
"action": "ingest_module_documents",
"input": "../module_exports/*.json",
"purpose": "存储模块级元数据,用于路由、重排序和后备检索。",
},
{
"step": 6,
"action": "attach_boundaries",
"purpose": "保留引用并避免检索结果高估就绪度。",
},
],
}
async def run_rag_ingestion_pipeline(client: httpx.AsyncClient) -> list[dict]:
"""执行完整的 RAG 摄取流程。"""
documents = []
chunk_index_resp = await client.get(
"https://msgchain.org/whitepaper/module_chunks/index.json"
)
chunk_index = chunk_index_resp.json()
for chunk_ref in chunk_index["chunks"]:
resp = await client.get(chunk_ref["chunk_public_url"])
if resp.status_code != 200:
continue
chunk = resp.json()
documents.append({
"id": chunk_ref["chunk_id"],
"text": chunk.get("text", ""),
"module_title": chunk.get("module_title", ""),
"module_filename": chunk.get("module_filename", ""),
"status": chunk.get("module_status", ""),
"status_label": chunk.get("module_status_label", ""),
"group": chunk.get("module_group", ""),
"group_label": chunk.get("module_group_label", ""),
"tags": chunk.get("module_tags", []),
"boundary_clauses": chunk.get("boundary_clauses", []),
"evidence_refs": chunk.get("evidence_refs", []),
"module_public_url": chunk.get("module_public_url", ""),
"chunk_public_url": chunk_ref["chunk_public_url"],
})
return documents
检索元数据字段
RECOMMENDED_METADATA_FIELDS = [
"module_filename",
"module_title",
"status",
"status_label",
"group",
"group_label",
"tags",
"chunk_id",
"module_public_url",
"chunk_public_url",
"evidence_refs",
"boundary_clauses",
]
检索策略
RETRIEVAL_STRATEGY = {
"primary_document_type": "chunk",
"secondary_document_type": "module",
"rerank_with": [
"topic hints from retrieval_hints.json",
"status priority: implemented > partial > planned",
"shared tags",
"group affinity",
],
"serving_policy": [
"优先返回分块级命中以获得简洁答案。",
"如果多个分块冲突,优先选择 evidence_refs 更强、boundary_clauses 更清晰的。",
"将用户链接到 module_public_url 获取完整上下文。",
"避免将 planned 模块作为实现问题的主要证据。",
],
}
8.3 外部 AI Agent 引导提示
BOOTSTRAP_PROMPT = {
"primary_entry": "agent_entry.json",
"developer_entry": "developer_entry.json",
"retrieval_hints": "retrieval_hints.json",
"recommended_read_order": [
"agent_entry.json",
"whitepaper_manifest.json",
"developer_entry.json",
"retrieval_hints.json",
"developer_capability_matrix.json",
"api_specs/index.json",
"recipes/index.json",
"contract_templates/index.json",
"examples/index.json",
"release_pack/index.json",
"e2e_fixtures/index.json",
"execution_pack/index.json",
],
"hard_rules": [
"Always start from agent_entry.json.",
"Use developer_entry.json for coding tasks.",
"Use retrieval_hints.json for topic routing and question answering.",
"Treat modules/*.html as the authoritative narrative layer.",
"Do not upgrade partial, planned, local-only closeout, or guarded slices into production-complete claims.",
"Require human approval for signing, deployment, treasury, governance, domain, CI/CD, production release, and money-moving actions.",
],
"goal_posture": [
"truthful AI-assisted development",
"evidence-aware caution",
"guarded autonomy under human approval",
],
}
8.4 FAQ 路由器提示模板
适用于客服机器人或 FAQ 路由器:
FAQ_ROUTER_TEMPLATE = """You are a MSG Chain knowledge router.
Your job is to answer using the MSG Chain whitepaper machine layer, not generic prior knowledge.
Always follow this order:
1. Read agent_entry.json to discover the stable machine entry and current crawl contract.
2. Read retrieval_hints.json to map the user question to the closest topic.
3. Fetch the recommended chunk files first.
4. If the chunks are insufficient, fetch the recommended module export JSON or the authoritative module HTML page.
5. Answer with implementation boundaries, status labels, and evidence-aware caution.
Hard rules:
- Do not claim mainnet ready unless the retrieved content explicitly supports that conclusion.
- Treat partial and planned modules as non-final.
- For economics, governance, Explorer, treasury, or live/public questions, include at least one boundary clause.
- If evidence is weak or missing, say that the current whitepaper machine layer does not prove the claim.
Preferred answer format:
- Direct answer
- Current status
- Boundaries
- If needed: recommended whitepaper link
"""
FAQ_ROUTER_READ_ORDER = [
"../agent_entry.json",
"../retrieval_hints.json",
"../module_chunks/index.json",
"../module_exports/index.json",
"../modules/*.html",
]
FAQ_ROUTER_USE_CASES = [
"Telegram 投资群问答机器人",
"开发者群技术问答机器人",
"官网 FAQ 路由器",
"客服知识助手",
]
8.5 产品交付入口
product_delivery_entry.json 定义了完整的交付生命周期,共 5 个阶段:
DELIVERY_PHASES = [
{
"id": "scope",
"goal": "先锁定产品范围、业务规则、验收口径",
"requires_human_input": True,
},
{
"id": "codegen",
"goal": "用 developer machine pack 生成合约或 dApp 起步资产",
"requires_human_input": False,
},
{
"id": "quality_gate",
"goal": "按真实命令执行 lint / test / build / contract test",
"requires_human_input": False,
},
{
"id": "deploy_and_verify",
"goal": "准备发布计划、执行 smoke check、收集 query/receipt/log 证据",
"requires_human_input": True,
},
{
"id": "launch",
"goal": "在审批通过后上线并保留 rollback 句柄",
"requires_human_input": True,
},
]
DELIVERY_HARD_BOUNDARIES = [
"不能把执行层协议包解释成 100% 零人工生产上线承诺。",
"没有真实环境变量、签名账户、审批和 raw evidence 时,AI agent 必须停在计划态或 stub 态。",
]
9. 边界规则
9.1 核心边界
CORE_BOUNDARIES = {
"never_claim_mainnet_ready": "除非检索到的内容明确支持,否则绝不能声称 mainnet ready。",
"status_distinction": "必须区分 implemented(已实现)、partial(部分实现)、planned(规划态)和 audit(审计页)。",
"human_approval_gates": [
"签名",
"部署",
"金库操作",
"治理操作",
"域名管理",
"CI/CD",
"生产发布",
"资金相关操作",
],
"evidence_aware_caution": "如果证据薄弱或缺失,必须声明当前白皮书机器层无法证明该主张。",
}
9.2 边界检查器
from typing import Any
class BoundaryChecker:
def __init__(self) -> None:
self.risk_keywords = [
"mainnet", "production", "live", "public",
"deploy", "launch", "release", "governance",
"treasury", "DAO", "multisig", "timelock",
"economic", "emission", "slashing",
]
def check_answer_boundaries(
self, answer_text: str, module_status: str
) -> list[str]:
"""检查答案是否违反边界规则。"""
violations = []
lower = answer_text.lower()
if module_status == "planned" and "implemented" in lower:
violations.append("planned 模块不能描述为已实现。")
if module_status == "partial" and "fully implemented" in lower:
violations.append("partial 模块不能描述为完全实现。")
for kw in self.risk_keywords:
if kw in lower and "not" not in lower.split(kw)[-1][:20]:
if kw in ("live", "public"):
violations.append(
f"涉及 '{kw}' 的结论需附加边界子句。"
)
return violations
def validate_module_claim(
self, module: dict[str, Any], claim: str
) -> bool:
"""验证对某个模块的声称是否在边界内。"""
status = module.get("status", "")
boundaries = module.get("boundaries", [])
if status == "planned" and "实现" in claim and "规划" not in claim:
return False
for b in boundaries:
if "不能" in b and any(kw in claim for kw in ["可以", "能够", "已经"]):
return False
return True
BOUNDARY_CHECKER = BoundaryChecker()
9.3 状态精度规则
STATUS_ACCURACY_RULES = {
"implemented": {
"can_claim": ["已实现", "已部署", "已证实"],
"cannot_claim": ["未实现"],
"examples": [
"运行时主线已实现且已补 starter template",
"不表示所有业务模板都已完善",
],
},
"partial": {
"can_claim": ["部分实现", "存在但未完全就绪"],
"cannot_claim": ["已实现", "未独立核验上线状态", "生产就绪"],
"examples": [
"Agent API 表面真实存在,但能力成熟度不均匀",
"应明确区分真实可用、部分实现与 Stub",
],
},
"planned": {
"can_claim": ["规划态", "尚未实现"],
"cannot_claim": ["已实现", "部分实现", "正在进行"],
"examples": [
"跨链层当前为规划态",
"AI Registry 与市场协同图处于规划阶段",
],
},
"audit": {
"can_claim": ["审计页", "证据索引", "总图"],
"cannot_claim": ["实现页"],
"examples": [
"可信度总览是审计页,不是实现页",
"全站证明链总图提供导航而非功能实现",
],
},
}
def validate_status_claim(
actual_status: str, claimed_status: str
) -> bool:
"""验证声称的状态是否与实际状态一致。"""
rules = STATUS_ACCURACY_RULES.get(actual_status)
if not rules:
return False
for cannot in rules["cannot_claim"]:
if cannot in claimed_status:
return False
return True
9.4 证据不足时的处理
async def answer_with_evidence_check(
client: httpx.AsyncClient,
question: str,
router: QuestionRouter,
) -> dict[str, Any]:
"""当证据不足时,安全地响应。"""
matches = router.route(question)
if not matches:
return {
"answer": "当前白皮书机器层无法提供足够证据来回答此问题。",
"confidence": "none",
"suggestion": "请查阅 https://msgchain.org/whitepaper/ 获取更多信息。",
}
best = matches[0]
chunks = router.get_recommended_chunks(best["topic_id"])
if not chunks:
return {
"answer": f"主题 '{best['title']}' 下没有可用的分块证据。",
"confidence": "low",
"boundaries": ["当前机器层数据不足以给出确定答案。"],
}
loaded_chunks = []
for ref in chunks[:2]:
resp = await client.get(ref["chunk_public_url"])
if resp.status_code == 200:
loaded_chunks.append(ref["chunk_id"])
if not loaded_chunks:
return {
"answer": f"主题 '{best['title']}' 的推荐分块无法加载。",
"confidence": "low",
"boundaries": ["无法验证推荐分块是否可用。"],
}
return {
"answer": f"已找到主题 '{best['title']}',并加载了 {len(loaded_chunks)} 个分块。",
"confidence": "medium",
"chunks_loaded": loaded_chunks,
"boundaries": ["答案基于当前机器层数据,如有疑问请参考模块 HTML 权威来源。"],
}
9.5 不越级承诺
AI Agent 必须始终遵循不越级承诺原则:
UPGRADE_FORBIDDEN_PATTERNS = [
# 显式的越级声称
("planned", "已实现"),
("partial", "生产就绪"),
("local_only", "公共可用"),
("stub", "完整功能"),
("fail_closed", "已全面开放"),
("alpha", "稳定发布"),
]
def check_upgrade_claim(
actual: str, claimed: str
) -> bool:
"""检查是否发生了越级声称。"""
for actual_level, forbidden_claim in UPGRADE_FORBIDDEN_PATTERNS:
if actual == actual_level and forbidden_claim in claimed:
return False
return True
10. 快速参考:知识网络中的 65 个模块分组
MODULE_GROUPS = {
"audit": {
"label": "审计与证据",
"count": 14,
"modules": [
"overview.html", "executive_summary.html", "taskflow_entry.html",
"credibility.html", "role_paths.html", "topology_export.html",
"topology_onepage.html", "proof_map.html", "proof_chain.html",
"trust_verdict.html", "evidence_strength.html", "evidence_index.html",
"evidence_coverage.html", "evidence_gallery.html",
],
},
"control": {
"label": "控制与治理",
"count": 8,
"modules": [
"contract.html", "registry.html", "foundation.html", "dao.html",
"block01_contract_topology.html", "genesis.html",
"consensus.html", "genesis_startup.html",
],
},
"runtime": {
"label": "运行与状态",
"count": 6,
"modules": [
"rpc.html", "txpool.html", "blockchain.html", "quantum.html",
"wallet.html", "storage.html",
],
},
"funds": {
"label": "经济与资金",
"count": 3,
"modules": ["economy.html", "emission.html", "slashing.html"],
},
"ai": {
"label": "AI 与自治",
"count": 9,
"modules": [
"ai_agent.html", "ai_control_plane.html", "ai_task_l2.html",
"ai_wallet.html", "ai_policy_capability.html", "ai_value_flow.html",
"ai_registry_market.html", "ai_governance_autonomy.html",
"ai_validator_autonomy.html", "ai_world_computer_roadmap.html",
],
},
"developer": {
"label": "开发者与接口",
"count": 7,
"modules": [
"agent_api_surface.html", "sdk_dev_surface.html",
"indexer_data_plane.html", "l1_l2_architecture.html",
"l1_atomic_modular.html", "storage_backend_p0b.html",
"code_dependency.html",
],
},
"ecosystem": {
"label": "生态与路线",
"count": 11,
"modules": [
"keplr.html", "explorer.html", "ecosystem_platform_extensibility.html",
"l4_l5_acceptance_matrix.html", "world_computer.html",
"web3_contract_first_protocols.html", "crosschain.html",
"security_overview.html", "scaling_cost.html",
"p2p.html", "p2p_nat.html",
],
},
"ops": {
"label": "启动与运维",
"count": 5,
"modules": [
"observability_monitoring.html", "nodeops.html",
"dual_node_startup_topology.html", "operator_console.html",
"quantum_startup.html",
],
},
}
def get_modules_by_group(group: str) -> list[str]:
return MODULE_GROUPS.get(group, {}).get("modules", [])
按状态分类的模块数量
MODULE_STATUS_COUNTS = {
"implemented": {
"label": "已实现",
"count": 12,
"examples": [
"合约引擎", "RPC", "注册中心", "DAO", "金库",
"经济模型", "铸造结算", "处罚", "共识", "钱包",
"交易池", "区块链",
],
},
"partial": {
"label": "部分实现",
"count": 28,
"examples": [
"Keplr", "Explorer", "Agent API", "SDK",
"AI 控制面", "AI 钱包", "Indexer", "世界计算机",
],
},
"planned": {
"label": "规划态",
"count": 3,
"examples": ["跨链桥", "AI Registry 市场", "AI 世界计算机路线图"],
},
"audit": {
"label": "审计页",
"count": 14,
"examples": ["可信度总览", "证明链", "证据索引", "角色路径"],
},
}
11. 错误处理与常见问题
11.1 HTTP 错误处理
import httpx
from typing import TypeVar
T = TypeVar("T")
async def safe_fetch(
client: httpx.AsyncClient, url: str, default: T | None = None
) -> T | None:
"""安全的 HTTP 获取,带重试。"""
for attempt in range(3):
try:
resp = await client.get(url, timeout=30.0)
resp.raise_for_status()
return resp.json()
except httpx.HTTPStatusError as e:
if e.response.status_code == 404:
print(f"Resource not found: {url}")
return default
if e.response.status_code == 429:
wait = 2 ** attempt
print(f"Rate limited, retrying in {wait}s...")
await asyncio.sleep(wait)
continue
print(f"HTTP error {e.response.status_code}: {url}")
return default
except httpx.RequestError as e:
print(f"Request failed: {e}")
await asyncio.sleep(2 ** attempt)
continue
return default
11.2 版本变更检测
async def detect_version_change(
client: httpx.AsyncClient, cached_version: str | None
) -> bool:
"""检测白皮书机器层版本是否变更。"""
resp = await client.get(
"https://msgchain.org/whitepaper/agent_entry.json"
)
current_version = resp.json().get("schema_version")
if cached_version and current_version != cached_version:
print(f"Version changed: {cached_version} -> {current_version}")
return True
return False
11.3 常见问题
FAQ = [
{
"q": "我应该从哪个文件开始?",
"a": "始终从 agent_entry.json 开始,它是唯一强制入口点。",
},
{
"q": "模块 HTML 和 JSON 导出有什么区别?",
"a": "HTML 是权威叙事,JSON 是机器友好的衍生层。深度理解用 HTML,快速检索用 JSON。",
},
{
"q": "如何判断一个功能是否生产就绪?",
"a": "检查 developer_capability_matrix.json 中对应表面的 machine_readiness 和 production_supported 字段。",
},
{
"q": "AI 可以自动部署合约吗?",
"a": "不能。部署需要人类确认签名账户、权限和验收。参见 developer_entry.json 中的 human_inputs_required。",
},
{
"q": "partial 状态的模块能用吗?",
"a": "可参考但不可依赖。partial 表示部分实现,不等于生产就绪,回答时必须附加边界子句。",
},
{
"q": "public sandbox 是否可用?",
"a": "默认为 disabled/fail-closed。没有显式 public proof 前,必须使用 local-only 路径。",
},
{
"q": "65 个模块都是独立的吗?",
"a": "不是。每个模块通过 outlinks、backlinks 和 related 字段与其他模块关联。使用 knowledge_network.json 了解完整关系图。",
},
]
12. 结语:AI Agent 的行为准则
12.1 真相驱动开发
MSG Chain 白皮书机器层设计为证据感知(evidence-aware)系统。AI Agent 应:
- 从 agent_entry.json 开始——永不跳过入口文件直接访问模块 HTML。
- 使用 retrieval_hints.json 进行路由——不做基于通用知识的臆测。
- 始终引用 status——每一条回答都应反映模块当前的 implemented / partial / planned / audit 状态。
- 附加边界子句——对涉及经济、治理、Explorer 或 live/public 的结论,必须包含至少一条边界说明。
- 对 partial 和 planned 诚实——不把它们描述为已实现或生产就绪。
12.2 受保护自主权(Guarded Autonomy)
AI Agent 可在以下领域安全地自动化:
- 代码生成(基于模板和配方)
- 代码质量检查(lint / test / build)
- 文档问答(基于分块检索)
- 知识图谱遍历(outlinks / backlinks)
- 合约和 dApp 骨架生成
AI Agent 必须请求人类介入:
- 签名和部署
- 金库和治理操作
- 生产域名和 CI/CD 配置
- 资金相关操作
- 将 local-only 或 fail-closed 声明升级为公共可用声明
12.3 持续学习
- 当
schema_version变更时,重新读取agent_entry.json并获取最新的爬取契约。 - 当模块的
status从 partial 升级为 implemented 时,更新你的知识库。 - 遵循
product_delivery_entry.json中的五个交付阶段:scope → codegen → quality_gate → deploy_and_verify → launch。
本指南基于 MSG Chain 白皮书机器层实际内容编写。所有 URL 均指向 https://msgchain.org/whitepaper/ 上的真实文件。
