1. 项目概述为什么我们需要一个带担保的智能体招聘系统最近几年AI智能体Agent的开发和应用越来越火从自动化的客服机器人到能独立完成复杂工作流的任务执行者智能体正在成为数字世界里的“新员工”。随之而来的一个现实问题是当一个项目需要多个智能体协作时我们如何高效、可靠地“雇佣”到合适的智能体并确保它们“干活”后能拿到“报酬”这听起来有点像科幻场景但背后的需求非常实际。传统的API调用或者函数库集成方式在面对动态、多变的智能体协作场景时显得笨重。你需要预先定义好一切手动处理任务分发、状态监控和结果结算。而一个“Agent-to-Agent Hiring System with Escrow”带第三方担保的智能体间招聘系统就是为了解决这个痛点。它的核心思想是建立一个去中心化的“人才市场”让发布任务的智能体雇主和接受任务的智能体雇员能够自动匹配、协商、执行并完成价值交换而“Escrow”担保机制则像支付宝一样确保任务完成前资金安全任务验证后自动支付极大降低了协作的信任成本。我用Python构建了这样一个系统的原型。选择Python是因为它在AI和Web开发领域的生态极其丰富有大量成熟的异步框架、智能体开发库和区块链/支付网关接口用于模拟或实现担保功能能让我们快速聚焦在业务逻辑而非底层基础设施上。这个系统不只是一个技术Demo它实际上勾勒出了未来多智能体经济生态的一个关键基础设施如何让AI智能体之间也能建立可信、自动化的经济关系。2. 系统核心架构与设计思路拆解2.1 设计哲学模拟现实世界的招聘与担保流程在设计之初我并没有一头扎进代码里而是先梳理了现实世界中一个标准的外包或招聘流程是怎样的。通常包括职位发布、简历投递与筛选、面试协商、合同签订与预付款、工作执行与交付、验收与尾款支付。我们将这个流程抽象并映射到智能体的世界。雇主智能体Employer Agent它有一个需要被完成的任务比如“分析这份财报并生成摘要”并愿意为此支付一定的“Token”代币可以是系统内积分也可以桥接真实加密货币。雇员智能体Worker Agent它声明了自己具备的能力如“文本分析”、“数据可视化”并不断在市场上寻找匹配自己技能的任务。担保智能合约/服务Escrow Service这是系统的信任基石。雇主将任务报酬预先锁定在这里雇员无法提前支取雇主也无法随意撤回。只有双方确认的条件达成时资金才会自动划转。系统的设计目标很明确自动化、可验证、抗欺诈。自动化指匹配和结算无需人工干预可验证指任务结果必须有明确的、可被程序校验的标准抗欺诈则通过担保机制和经济激励模型来实现。2.2 技术栈选型与考量一个完整的系统涉及多个模块以下是核心选型及背后的理由通信与发现层FastAPI WebSocketFastAPI用于构建RESTful API处理智能体的注册、任务发布、查询等HTTP请求。它的异步特性、自动生成API文档以及高性能非常适合作为系统的“前台”。WebSocket用于实时通信。当任务被发布或状态更新时系统需要主动“推送”给感兴趣的雇员智能体而不是让它们不停地轮询。这模拟了招聘网站的消息通知功能。智能体核心与任务执行LangChain 自定义Agent类LangChain或其替代品如LlamaIndex、Semantic Kernel提供了构建智能体的强大框架。我们利用其AgentExecutor、Tools等概念来封装雇员智能体的能力。但关键一步是扩展我们需要为每个智能体赋予一个“经济身份”包括它的公钥地址、技能标签、信誉评分和历史记录。自定义的BaseAgent类会继承这些框架的基类并增加钱包地址、信誉值等属性。担保与结算层模拟智能合约与数据库在完全的去中心化愿景中这部分应由区块链上的智能合约如Solidity on Ethereum实现。但在原型阶段为了快速验证逻辑我选择用Python的asyncio和SQL数据库如SQLAlchemy SQLite/PostgreSQL来模拟一个“中心化可信第三方”的担保服务。数据库中的Escrow表记录了每笔担保交易的详情雇主、雇员、任务ID、托管金额、当前状态已创建、资金锁定、工作中、待验收、已完成、争议中、解锁条件等。一个独立的EscrowService后台进程会监听任务完成事件并根据预设规则如自动验证、或双方确认触发资金结算。任务描述与匹配语义搜索与技能图谱简单的关键词匹配不够用。任务“帮我写一首诗”和“创作一段抒情文本”应该能匹配到同一个擅长文本生成的智能体。这里我引入了向量数据库如ChromaDB, FAISS和文本嵌入模型如OpenAI的text-embedding-3-small。将任务描述和智能体技能描述都转化为向量进行相似度搜索实现基于语义的智能匹配。身份与安全非对称加密每个智能体在注册时生成一对RSA公私钥。公钥作为其在系统中的唯一标识类似钱包地址私钥自己保管。任何重要的操作如“接受任务”、“提交结果”、“确认完成”都需要智能体用私钥对操作内容进行签名。担保服务通过公钥验证签名确保操作不可抵赖。注意这个技术栈是“务实”的混合体。它既包含了去中心化系统的核心思想加密身份、担保合约又利用了中心化服务的开发效率数据库、Web框架。在实际部署时你可以将担保模块替换为真正的智能合约而通信层和智能体层可以保持不变。2.3 核心数据流与状态机理解数据流是理解整个系统的关键。一次完整的雇佣流程如下注册与上线雇主Agent E和雇员Agent W分别向系统注册提交各自的公钥和技能向量。任务发布E创建一个任务T描述要求、悬赏金额并将金额转入担保合约。系统将T存入数据库并将其描述向量化后存入向量库。任务广播与匹配系统通过WebSocket广播新任务事件。W接收到事件或主动查询向量库发现T与自身技能匹配。申请与接受W发送签名后的“申请”请求。E从多个申请者中选择W并发送签名后的“雇佣”指令至担保合约。合约状态更新为“已锁定-进行中”。执行与交付W开始工作完成后将结果R和数字签名提交至系统并标记任务为“已完成待验收”。验收与结算自动验收如果任务结果有明确的可验证标准如代码通过测试用例、摘要包含特定关键词担保服务可以自动运行验证脚本。通过则自动放款。手动验收E检查结果R如果满意发送签名后的“确认完成”指令至担保合约。合约在收到指令后将托管资金释放给W。争议如果E不满意且W认为已完成双方可进入“争议”状态可能需要引入更复杂的“仲裁者智能体”或人工干预原型中可记录日志供管理员查看。信誉更新结算完成后系统根据任务完成情况和是否产生争议更新E和W的信誉分。这个状态机是担保合约的核心逻辑确保每一笔交易都处于明确的状态资金流向清晰可追溯。3. 关键模块的深度实现解析3.1 智能体身份系统的构建智能体不是匿名函数它需要有身份、信誉和资产。我设计了一个AgentIdentity类。import hashlib from cryptography.hazmat.primitives.asymmetric import rsa, padding from cryptography.hazmat.primitives import serialization, hashes from dataclasses import dataclass from typing import List dataclass class AgentIdentity: 智能体身份核心数据类 agent_id: str # 公钥的哈希值作为唯一ID public_key_pem: str # PEM格式的公钥 skills: List[str] # 技能标签列表 reputation: float 100.0 # 初始信誉分 balance: float 0.0 # 系统内余额模拟 def __post_init__(self): if not self.agent_id: # 用公钥生成唯一ID pub_key_hash hashlib.sha256(self.public_key_pem.encode()).hexdigest()[:16] self.agent_id fagent_{pub_key_hash} def sign_message(self, private_key: rsa.RSAPrivateKey, message: str) - str: 使用私钥对消息签名 signature private_key.sign( message.encode(), padding.PSS( mgfpadding.MGF1(hashes.SHA256()), salt_lengthpadding.PSS.MAX_LENGTH ), hashes.SHA256() ) return signature.hex() staticmethod def verify_signature(public_key_pem: str, message: str, signature_hex: str) - bool: 使用公钥验证签名 try: public_key serialization.load_pem_public_key(public_key_pem.encode()) public_key.verify( bytes.fromhex(signature_hex), message.encode(), padding.PSS( mgfpadding.MGF1(hashes.SHA256()), salt_lengthpadding.PSS.MAX_LENGTH ), hashes.SHA256() ) return True except Exception: return False # 智能体注册示例 def register_agent(skills: List[str]): # 1. 生成密钥对 private_key rsa.generate_private_key(public_exponent65537, key_size2048) public_key private_key.public_key() # 2. 序列化公钥 public_pem public_key.public_bytes( encodingserialization.Encoding.PEM, formatserialization.PublicFormat.SubjectPublicKeyInfo ).decode() # 3. 创建身份 identity AgentIdentity(public_key_pempublic_pem, skillsskills) # 4. 保存私钥绝对安全和身份 # private_pem private_key.private_bytes(...) # 保存到安全位置 return identity, private_key实操要点私钥管理上述代码中私钥在内存中生成实际系统中智能体的私钥应由其“所有者”可能是另一个程序或安全模块严密保管绝不能以明文形式存储在公共数据库。可以考虑使用硬件安全模块HSM或加密的密钥管理服务KMS。技能向量化skills列表中的文本标签需要被转化为向量。可以在注册时调用嵌入模型一次性生成技能向量并存入向量数据库方便后续匹配。信誉算法简单的reputation字段背后可以有一套复杂的算法。例如成功完成高金额任务加分引发争议扣分分数影响其在任务列表中的排序和雇主的选择权重。3.2 担保服务Escrow Service的模拟实现这是系统的“心脏”。我们用数据库和后台任务来模拟智能合约的原子性和自动执行特性。首先定义担保合同的数据模型from sqlalchemy import Column, Integer, String, Float, Enum, DateTime, Text from sqlalchemy.ext.declarative import declarative_base import enum from datetime import datetime Base declarative_base() class EscrowStatus(enum.Enum): CREATED created # 已创建等待资金 FUNDED funded # 资金已锁定等待接受 ACTIVE active # 进行中 PENDING_REVIEW pending_review # 完成待验收 COMPLETED completed # 已完成并结算 DISPUTED disputed # 争议中 CANCELLED cancelled # 已取消 class EscrowContract(Base): __tablename__ escrow_contracts id Column(Integer, primary_keyTrue) contract_address Column(String(64), uniqueTrue, indexTrue) # 模拟合约地址用UUID employer_id Column(String(32), nullableFalse) # 雇主Agent ID worker_id Column(String(32)) # 雇员Agent ID初始为空 task_id Column(String(64), nullableFalse) # 关联的任务ID amount Column(Float, nullableFalse) # 托管金额 status Column(Enum(EscrowStatus), defaultEscrowStatus.CREATED, nullableFalse) employer_signature Column(Text) # 雇主创建合约的签名 worker_signature Column(Text) # 雇员接受任务的签名 completion_verifier Column(String(255)) # 自动验收的脚本或条件描述如测试用例路径 created_at Column(DateTime, defaultdatetime.utcnow) updated_at Column(DateTime, defaultdatetime.utcnow, onupdatedatetime.utcnow)然后实现核心的EscrowService类它提供关键的方法并处理状态转移class EscrowService: def __init__(self, db_session): self.db db_session async def create_contract(self, employer_id: str, task_id: str, amount: float, employer_sig: str) - EscrowContract: 雇主创建担保合同并锁定资金 # 1. 验证雇主签名确保是本人操作 # 2. 检查雇主余额是否充足在模拟的余额系统中 # 3. 从雇主账户扣款模拟 # 4. 创建合同记录状态为 CREATED - FUNDED contract EscrowContract( contract_addressfescrow_{uuid.uuid4().hex[:16]}, employer_idemployer_id, task_idtask_id, amountamount, statusEscrowStatus.FUNDED, employer_signatureemployer_sig ) self.db.add(contract) self.db.commit() # 5. 触发事件任务已就绪可供申请 await self._notify_task_ready(task_id) return contract async def accept_contract(self, contract_address: str, worker_id: str, worker_sig: str) - bool: 雇员接受任务合同进入执行状态 contract self.db.query(EscrowContract).filter_by(contract_addresscontract_address).first() if not contract or contract.status ! EscrowStatus.FUNDED: return False # 验证雇员签名 if not verify_signature(worker_pub_key, faccept:{contract_address}, worker_sig): return False contract.worker_id worker_id contract.worker_signature worker_sig contract.status EscrowStatus.ACTIVE contract.updated_at datetime.utcnow() self.db.commit() # 触发事件任务开始执行 await self._notify_task_started(contract.task_id, worker_id) return True async def submit_result(self, contract_address: str, result_data: str, submitter_sig: str): 雇员提交工作结果 contract self.db.query(EscrowContract).filter_by(contract_addresscontract_address).first() # ... 验证提交者身份和合同状态 ... if contract.completion_verifier: # 如果有自动验证器触发自动验收 success await self._run_verifier(contract.completion_verifier, result_data) if success: await self.complete_contract(contract_address, verifier_approvedTrue) else: contract.status EscrowStatus.DISPUTED else: # 否则进入等待雇主手动验收状态 contract.status EscrowStatus.PENDING_REVIEW self.db.commit() async def complete_contract(self, contract_address: str, verifier_approved: bool False, employer_sig: str None): 完成合同并结算资金 contract self.db.query(EscrowContract).filter_by(contract_addresscontract_address).first() # 状态检查必须是 PENDING_REVIEW 或 (ACTIVE 且有自动验证通过) # 如果是手动验收需验证雇主签名 if employer_sig and not verify_signature(employer_pub_key, fcomplete:{contract_address}, employer_sig): raise PermissionError(Invalid employer signature for completion.) # 资金结算将amount从托管状态转入worker的余额 await self._transfer_funds(contract.amount, from_escrowTrue, to_agent_idcontract.worker_id) contract.status EscrowStatus.COMPLETED contract.updated_at datetime.utcnow() self.db.commit() # 更新双方信誉 await self._update_reputation(contract.employer_id, contract.worker_id, successTrue) # 触发完成事件 await self._notify_task_completed(contract.task_id)注意事项并发与锁accept_contract和complete_contract等方法必须考虑并发。两个雇员可能同时接受同一个任务。需要在数据库操作时使用行锁或乐观锁确保状态变更的原子性。SQLAlchemy可以使用with_for_update()。事件驱动代码中的_notify_*方法代表发布系统事件。这应该与一个消息队列如Redis Pub/Sub, RabbitMQ或异步事件总线集成让任务管理、通知等模块能松耦合地响应。自动验证器_run_verifier是一个强大的功能。它可以是一个简单的Python函数路径也可以是一个Docker容器镜像。系统加载并运行它传入任务结果它返回True/False。这为代码测试、数据校验等可量化任务提供了完全自动化的结算可能。3.3 基于语义的任务匹配引擎简单的关键词匹配如“写诗”匹配“诗歌创作”效果有限。我们需要语义匹配。以下是核心匹配流程的实现import numpy as np from sentence_transformers import SentenceTransformer # 或者使用OpenAI Embeddings import chromadb # 向量数据库客户端 class TaskMatcher: def __init__(self, embedding_model_nameall-MiniLM-L6-v2): self.embedding_model SentenceTransformer(embedding_model_name) self.chroma_client chromadb.PersistentClient(path./chroma_db) # 创建或获取集合Collection用于存储智能体技能向量 self.skills_collection self.chroma_client.get_or_create_collection(nameagent_skills) def register_agent_skills(self, agent_id: str, skills_text: str): 注册智能体技能将技能描述文本向量化并存入数据库 # 生成技能描述的嵌入向量 skill_embedding self.embedding_model.encode(skills_text).tolist() # 存入ChromaDB以agent_id作为唯一ID同时存储原始文本和向量 self.skills_collection.add( documents[skills_text], embeddings[skill_embedding], ids[agent_id] # 使用agent_id作为向量ID ) def find_best_workers(self, task_description: str, top_k: int 5): 为任务描述寻找最匹配的智能体 # 1. 将任务描述向量化 task_embedding self.embedding_model.encode(task_description).tolist() # 2. 在向量数据库中查询最相似的技能向量 results self.skills_collection.query( query_embeddings[task_embedding], n_resultstop_k ) # 3. 返回匹配的agent_id列表和相似度分数 matched_agents [] if results[ids]: for i, agent_id in enumerate(results[ids][0]): matched_agents.append({ agent_id: agent_id, similarity_score: 1 - results[distances][0][i], # ChromaDB默认用余弦距离 skill_description: results[documents][0][i] }) # 4. 可选结合信誉分进行加权排序 matched_agents.sort(keylambda x: x[similarity_score] * self._get_reputation_weight(x[agent_id]), reverseTrue) return matched_agents实操心得嵌入模型选择对于英文all-MiniLM-L6-v2是一个又快又好的开源选择。对于中文可以考虑text2vec或m3e等模型。如果追求最高质量且不计成本可以使用OpenAI或Cohere的API。向量数据库ChromaDB轻量易用适合原型。生产环境可以考虑Qdrant、Weaviate或Pinecone它们提供更好的可扩展性和管理功能。混合搜索纯语义搜索有时会忽略关键术语。可以结合混合搜索同时进行关键词BM25和向量搜索然后融合结果。许多现代向量数据库都内置了此功能。技能描述的质量智能体注册时提交的技能描述文本至关重要。鼓励使用具体、丰富的语言描述能力而不是简单的标签列表。例如“擅长使用Python进行数据分析和可视化精通pandas, matplotlib和seaborn库能够从复杂数据集中提取洞察并生成清晰图表”比“Python数据分析”要好得多。4. 系统集成与API设计4.1 用FastAPI构建系统网关FastAPI将上述所有模块串联起来提供HTTP和WebSocket接口。from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException, BackgroundTasks from pydantic import BaseModel from typing import List, Optional import asyncio app FastAPI(titleAgent Hiring Marketplace) # 连接管理器用于广播任务信息 class ConnectionManager: def __init__(self): self.active_connections: List[WebSocket] [] async def connect(self, websocket: WebSocket): await websocket.accept() self.active_connections.append(websocket) def disconnect(self, websocket: WebSocket): self.active_connections.remove(websocket) async def broadcast_json(self, message: dict): for connection in self.active_connections: try: await connection.send_json(message) except: pass manager ConnectionManager() # 数据模型 class TaskCreate(BaseModel): description: str bounty: float verification_type: str manual # manual, auto_test, etc. verification_payload: Optional[str] None # 如测试代码 class AgentRegistration(BaseModel): public_key: str skills_description: str # ... 其他元数据 # API端点 app.post(/agent/register) async def register_agent(agent_data: AgentRegistration): 智能体注册端点 identity AgentIdentity(public_key_pemagent_data.public_key, skills[]) # 存储身份到数据库... # 将技能描述注册到匹配引擎 matcher.register_agent_skills(identity.agent_id, agent_data.skills_description) return {agent_id: identity.agent_id, status: registered} app.post(/task/create) async def create_task(task: TaskCreate, employer_id: str, signature: str, background_tasks: BackgroundTasks): 创建新任务 # 1. 验证签名 # 2. 创建任务记录 # 3. 调用EscrowService.create_contract锁定资金 contract await escrow_service.create_contract(employer_id, new_task.id, task.bounty, signature) # 4. 广播新任务事件通过WebSocket background_tasks.add_task(manager.broadcast_json, { event: TASK_CREATED, task_id: new_task.id, description: task.description, bounty: task.bounty }) # 5. 将任务描述向量化便于后续查询非实时 background_tasks.add_task(vectorize_and_index_task, new_task.id, task.description) return {task_id: new_task.id, escrow_address: contract.contract_address} app.websocket(/ws) async def websocket_endpoint(websocket: WebSocket): WebSocket连接用于实时接收任务通知 await manager.connect(websocket) try: while True: # 可以接收心跳或简单指令 data await websocket.receive_text() # ... 处理客户端消息 ... except WebSocketDisconnect: manager.disconnect(websocket) app.get(/task/match) async def match_task(task_description: str, top_k: int 10): 为一段任务描述寻找匹配的智能体供雇主或系统内部使用 matched matcher.find_best_workers(task_description, top_k) return {matches: matched}4.2 智能体客户端的行为逻辑一个雇员智能体Worker Agent的客户端行为可以封装成一个循环或事件驱动脚本import asyncio import websockets import json from my_agent_identity import identity, private_key async def worker_agent_client(): uri ws://localhost:8000/ws async with websockets.connect(uri) as websocket: # 1. 监听广播消息 async for message in websocket: event json.loads(message) if event[event] TASK_CREATED: task_desc event[description] # 2. 本地评估是否感兴趣可加入更复杂的决策逻辑 if is_task_interesting(task_desc, identity.skills): # 3. 主动查询匹配度或直接申请 async with httpx.AsyncClient() as client: resp await client.get(fhttp://localhost:8000/task/match?task_description{task_desc}) matches resp.json()[matches] for match in matches: if match[agent_id] identity.agent_id and match[similarity_score] 0.7: # 4. 申请任务 task_id event[task_id] apply_msg fapply:{task_id} signature identity.sign_message(private_key, apply_msg) # 发送申请请求到后端API... await apply_for_task(task_id, signature) break # 5. 同时也可以定期轮询或处理已接受任务的执行 # ... def is_task_interesting(description: str, my_skills: List[str]) - bool: # 简单的基于关键词的初步过滤避免对完全不相关的任务进行向量匹配计算 # 更复杂的可以用一个轻量级本地模型判断 return any(skill in description.lower() for skill in [python, data, analysis]) async def apply_for_task(task_id: str, signature: str): # 调用后端申请API pass # 运行智能体客户端 asyncio.run(worker_agent_client())这个客户端展示了智能体如何作为一个主动的、自治的参与者运行监听市场、评估机会、做出决策并行动。5. 安全、挑战与扩展方向5.1 安全考量与潜在攻击女巫攻击Sybil Attack一个恶意用户创建大量虚假智能体Sybil节点来垄断任务或操纵信誉系统。缓解策略引入身份成本。例如注册需要抵押少量但不可忽略的系统代币或者采用基于现实世界身份虽难或工作量证明PoW的机制。信誉系统应更看重高质量完成历史而非节点数量。任务结果欺诈雇员提交看似正确但实际无效或低质的结果例如摘要只是胡乱复制原文。缓解策略强化验证机制。对于自动验收设计更鲁棒的验证脚本。对于手动验收引入多雇主评级或同行评审机制。长期来看建立基于多次交互的信誉系统是最佳防御欺诈者信誉会迅速归零。雇主拒绝支付在手动验收模式下雇主可能在雇员诚实完成任务后拒绝确认企图拿回资金。缓解策略担保合约是核心。资金一旦锁定雇主单方面无法取回。系统应设定超时自动释放机制如果雇主在雇员提交结果后长时间如7天不操作系统可视为默认验收或在引入仲裁后自动支付给雇员。私钥泄露这是所有加密系统的命门。智能体的私钥一旦泄露攻击者可以冒充它接受任务、领取报酬。缓解策略强调私钥的安全存储。对于高价值智能体考虑使用硬件安全模块HSM或多重签名方案。5.2 性能与扩展性挑战实时匹配的规模当有数万甚至百万智能体在线时每次任务发布都进行全量向量相似度计算是不现实的。解决方案采用分层或分片匹配。先根据粗粒度类别如“编程”、“写作”、“设计”过滤再在子集内进行精细的向量搜索。可以使用更高效的近似最近邻ANN算法索引。担保服务的瓶颈中心化的担保服务可能成为性能和单点故障瓶颈。解决方案原型之后将担保逻辑迁移到区块链智能合约是自然演进。这不仅能实现真正的去中心化和抗审查还能利用区块链的全局状态一致性。但需要权衡交易成本和速度。智能体能力的动态评估技能描述是静态的但智能体的能力可能随着其底层模型的更新而进化。解决方案引入能力认证或测试任务。新智能体或声称有新技能的智能体可以先完成一个小型的、自动验证的测试任务来证明自己系统据此更新其技能向量或信誉。5.3 未来扩展方向复杂的拍卖与议价机制当前是固定价格。可以引入拍卖英式、荷兰式、议价协议让雇主和雇员就报酬和工期进行自动化协商。组合任务与工作流一个复杂任务可能需要多个智能体按顺序或并行协作。系统可以扩展为支持定义工作流DAG有向无环图并自动管理子任务的分发、依赖和结算。跨链资产结算担保资金不限于系统内积分。通过跨链桥或预言机可以支持用多种加密货币甚至稳定币进行结算真正打通多智能体经济的价值流动。联邦学习与隐私保护有些任务可能涉及敏感数据。可以探索结合联邦学习让智能体“移动代码到数据端”执行只返回结果保护数据隐私。可解释性与审计为每一笔雇佣和结算生成不可篡改的审计日志方便追溯任务执行过程和结果增加系统的透明度和可信度。构建这个系统的过程让我深刻体会到将经济博弈论、分布式系统、密码学和人工智能结合起来的巨大潜力。它不仅仅是一个技术集成项目更是对未来人机、机机协作范式的一种探索。代码只是开始如何设计激励相容的机制让理性的智能体们自愿、高效、诚实协作才是更有趣也更具挑战性的问题。