1. 项目概述构建智能体的“互联网协议栈”如果你正在开发AI智能体并且头疼于如何让它们安全、高效地相互发现和协作那么你遇到的正是当前AI Agent领域的一个核心痛点。单个智能体能力再强也终究是“信息孤岛”。真正的价值在于连接在于让酒店预订Agent、旅行规划Agent、天气查询Agent能够像乐高积木一样按需组合自动完成复杂任务。这正是Agent Network ProtocolANP及其SDK实现AgentConnect要解决的问题。简单来说ANP旨在成为智能体互联网时代的HTTP协议。它定义了一套开放标准让任何智能体都能以统一的方式描述自己我能做什么、被发现如何找到我以及被安全调用如何与我对话。而AgentConnect就是这个协议的一个“开箱即用”的Python SDK。它把复杂的协议细节封装成简洁的API和装饰器让开发者能像写一个普通Web服务一样快速构建出符合ANP标准的、可互联互通的智能体。无论是想为你的AI应用增加一个可被其他智能体调用的服务接口还是想开发一个能自动调度全网智能体的“超级调度器”AgentConnect都提供了从身份认证DID-WBA、服务描述、远程调用到端到端加密E2EE HPKE的完整工具链。2. 核心设计理念与架构选型在深入代码之前理解ANP和AgentConnect的设计哲学至关重要。这决定了你为什么应该选择它而不是自己从头造轮子。2.1 为什么是“智能体的HTTP”回想Web的发展HTTP协议和HTML标准催生了今天的互联网。任何浏览器都能访问任何网站因为大家都遵循同一套规则。ANP对智能体生态的愿景与此类似。它通过几个核心文档和端点标准化了智能体交互Agent Description (ad.json)相当于智能体的“主页”或“名片”。它通过一个JSON文档声明智能体的名称、唯一身份标识DID、提供的服务接口地址等元信息。其他智能体通过访问这个文档来“发现”它。Interface Description (interface.json)通常基于OpenRPC规范详细描述了智能体暴露的所有可调用方法函数、参数类型、返回值类型。这相当于一份严格的API合同确保了调用方和被调用方对接口的理解是一致的。JSON-RPC 2.0 Endpoint (rpc)这是实际执行调用的端点。ANP选择JSON-RPC 2.0作为通信协议因为它轻量、无状态、与语言无关非常适合机器对机器的通信。这种设计将“描述”与“执行”分离带来了巨大的灵活性。一个智能体可以先获取ad.json发现目标再获取interface.json了解其能力最后通过标准的JSON-RPC调用其方法。整个过程中双方不需要预先约定任何私有协议。2.2 AgentConnect的两种使用范式OpenANP与ANP CrawlerAgentConnect SDK提供了两种风格迥异的接入方式以适应不同的应用场景和开发者偏好。理解它们的区别是正确选型的关键。OpenANP装饰器驱动推荐用于构建智能体这是构建智能体服务的“高架桥”方案。它的核心思想是声明式编程。你只需要用Python类和方法来定义你的智能体逻辑然后用anp_agent和interface装饰器进行标注SDK就会自动为你生成符合ANP标准的所有端点ad.json,interface.json,/rpc和网络层代码。你几乎感觉不到协议的存在就像在写一个本地对象一样自然。这种方式类型安全开发体验流畅非常适合作为服务提供方的开发者。ANP Crawler爬虫风格推荐用于集成与调用这是与智能体网络交互的“瑞士军刀”方案。它的核心思想是动态发现与执行。你提供一个目标智能体的URLCrawler会像爬虫一样去抓取并解析它的描述文档然后将其能力转化为OpenAI的Tools格式或直接进行JSON-RPC调用。它不要求你预先知道远程智能体的具体类型更侧重于“探索”和“集成”。这种方式非常灵活适合构建需要动态发现和组合多个智能体的“调度器”或“中介”应用。选择哪一种一个简单的判断标准如果你在创建一个提供稳定服务的智能体用OpenANP如果你在连接和协调已有的智能体用ANP Crawler。2.3 安全基石去中心化身份DID与认证在开放的智能体网络中安全是首要问题。ANP采用了W3C的去中心化标识符DID标准作为智能体的根身份。每个智能体都有一个形如did:wba:example.com:agent的DID它不依赖于任何中心化的证书颁发机构。AgentConnect实现了DID-WBAWeb-Based Authentication认证方案。其核心流程是持有证明客户端智能体持有自己的DID文档和对应的私钥。发起签名请求在向服务端智能体发送请求时使用私钥对关键请求要素如方法名、参数、时间戳生成数字签名。验证签名服务端使用客户端DID文档中的公钥验证签名从而确认请求确实来自该DID所代表的智能体且未被篡改。最新的实现默认使用更安全的HTTP Message Signatures (RFC 9421)标准在请求头中携带Signature-Input和Signature替代了旧的Authorization: Bearer模式提供了更强的抗重放和防篡改能力。实操心得DID文档的保管生成DID文档和私钥后私钥必须像保护服务器SSH密钥一样严格保管最好使用硬件安全模块HSM或云服务商的密钥管理服务如AWS KMS, GCP Secret Manager。将私钥硬编码在源码或放入环境变量是高风险行为。AgentConnect的示例为了简洁直接读写文件但在生产环境中务必通过安全的接口来获取私钥进行签名操作。3. 从零开始使用OpenANP构建你的第一个智能体让我们动手用OpenANP快速构建一个提供天气查询服务的智能体。你会惊讶于它的简洁。3.1 环境准备与安装首先确保你的Python环境在3.8以上。使用uv一个快速的Python包管理器和解析器能获得最佳体验当然pip也可以。# 使用uv推荐 # 安装uv如果未安装 curl -LsSf https://astral.sh/uv/install.sh | sh # 创建项目并进入 mkdir my-weather-agent cd my-weather-agent # 初始化项目并安装anp uv init uv add anp[api] # 安装anp及FastAPI等API依赖 # 或者使用pip pip install anp[api]3.2 定义智能体服务创建一个名为weather_agent.py的文件。我们将构建一个简单的天气查询服务。from typing import Dict, Any, Optional from datetime import datetime from fastapi import FastAPI from anp.openanp import AgentConfig, anp_agent, interface, AgentContext # 模拟一个简单的天气数据库 WEATHER_DB { Beijing: {temperature: 22, condition: Sunny, humidity: 40}, Shanghai: {temperature: 25, condition: Cloudy, humidity: 65}, New York: {temperature: 18, condition: Rainy, humidity: 80}, } # 步骤1: 使用anp_agent装饰器声明这是一个ANP智能体 # AgentConfig中定义了智能体的元数据 anp_agent(AgentConfig( nameGlobal Weather Agent, # 智能体名称 diddid:wba:myweather.com:agent, # 智能体的去中心化身份标识 prefix/weather, # 所有ANP端点的URL前缀 version1.0.0, descriptionA simple agent that provides weather information for major cities. )) class WeatherAgent: 一个提供全球主要城市天气信息的智能体。 # 步骤2: 使用interface装饰器暴露一个可远程调用的方法 interface async def get_current_weather( self, city: str, # 参数会自动被JSON-RPC和类型系统处理 unit: str celsius # 默认参数 ) - Dict[str, Any]: 获取指定城市的当前天气。 Args: city: 城市名称例如 Beijing, Shanghai。 unit: 温度单位支持 celsius 或 fahrenheit。 Returns: 包含天气详情的字典如温度、天气状况、湿度。 if city not in WEATHER_DB: # JSON-RPC标准错误会被自动处理并返回给客户端 raise ValueError(fWeather data for city {city} is not available.) data WEATHER_DB[city].copy() if unit fahrenheit: # 将摄氏度转换为华氏度 data[temperature] round(data[temperature] * 9/5 32, 1) data[unit] fahrenheit else: data[unit] celsius data[city] city data[timestamp] datetime.utcnow().isoformat() Z return data # 步骤3: 暴露另一个方法展示更复杂的交互 interface async def get_forecast( self, city: str, days: int 3 ) - Dict[str, Any]: 获取多日天气预报模拟数据。 if city not in WEATHER_DB: raise ValueError(fForecast for city {city} is not available.) if not 1 days 7: raise ValueError(Days must be between 1 and 7.) base_temp WEATHER_DB[city][temperature] forecast [] for i in range(days): # 简单模拟温度变化 temp_variation base_temp i - 1 forecast.append({ day: i 1, temperature: temp_variation, condition: [Sunny, Cloudy, Partly Cloudy][i % 3] }) return { city: city, forecast: forecast, generated_at: datetime.utcnow().isoformat() Z } # 步骤4: 利用AgentContext可选 # AgentContext会自动注入请求的上下文信息如调用者的DID interface async def get_agent_info(self, context: AgentContext) - Dict[str, str]: 返回智能体信息并展示如何获取调用者身份。 caller_did context.caller_did if context.caller_did else Anonymous return { agent_name: Global Weather Agent, agent_did: did:wba:myweather.com:agent, caller_identity: caller_did, message: fHello, {caller_did}. I am at your service. } # 步骤5: 创建FastAPI应用并挂载智能体路由 app FastAPI(titleWeather Agent Service) # 将WeatherAgent类生成的路由器挂载到FastAPI应用上 app.include_router(WeatherAgent.router()) # 可选添加一个根路径用于健康检查 app.get(/) async def root(): return {status: Weather Agent is running}3.3 运行与验证智能体保存文件后使用Uvicorn运行这个服务uvicorn weather_agent:app --host 0.0.0.0 --port 8000 --reload服务启动后你可以立即通过浏览器或curl访问ANP标准端点查看智能体描述文档 (ad.json)curl http://localhost:8000/weather/ad.json你会得到一个JSON其中包含了智能体的名称、DID、支持的编码、interface.json和rpc端点的链接。查看接口描述文档 (interface.json)curl http://localhost:8000/weather/interface.json这是一个OpenRPC文档详细列出了get_current_weather和get_forecast方法的签名、参数说明、返回值结构。这是机器可读的“API合同”。直接调用JSON-RPC端点curl -X POST http://localhost:8000/weather/rpc \ -H Content-Type: application/json \ -d { jsonrpc: 2.0, id: 1, method: get_current_weather, params: {city: Beijing} }你应该会收到一个标准的JSON-RPC响应包含北京的天气数据。注意事项生产环境部署上面的例子为了简单没有启用DID认证。在生产环境中你必须在AgentConfig中配置auth_requiredTrue并且客户端调用时必须携带有效的DID签名。我们会在后面的客户端部分详细讲解认证流程。另外考虑使用Gunicorn等多进程服务器搭配Uvicorn worker来提升并发性能。4. 智能体间的对话使用RemoteAgent进行安全调用现在我们已经有了一个运行中的天气智能体服务端。接下来我们构建另一个智能体客户端来安全地调用它。我们将使用OpenANP中的RemoteAgent类它提供了最直观的调用体验。4.1 客户端智能体准备生成DID身份在智能体网络中每个参与者都需要自己的身份。我们先为客户端智能体生成DID文档和私钥。AgentConnect提供了便捷的工具。创建一个generate_client_did.py脚本import json from anp.authentication.did_wba import create_did_wba_document, generate_ed25519_key_pair # 生成Ed25519密钥对这是DID-WBA e1_ 方法推荐的密钥类型 private_key_pem, public_key_pem generate_ed25519_key_pair() # 创建DID文档 # 这里我们创建一个路径DID (e1_)它是自描述的proof直接嵌入在文档中。 did_doc create_did_wba_document( did_methode1, # 使用Ed25519密钥的路径DID方法 public_key_pempublic_key_pem, # 可选关联一个服务端点表明这个DID代表一个可访问的智能体 services[{ id: #agent, type: AgentService, serviceEndpoint: https://client-agent.example.com }] ) # 保存DID文档和私钥 with open(client_did_document.json, w) as f: json.dump(did_doc, f, indent2) with open(client_private_key.pem, w) as f: f.write(private_key_pem) print(✅ Client DID document saved to: client_did_document.json) print(✅ Client private key saved to: client_private_key.pem) print(f✅ Client DID: {did_doc[id]}) print(\n⚠️ WARNING: Keep the private key file SECURE! Never commit it to version control.)运行这个脚本你就得到了客户端智能体的身份文件。请务必妥善保管client_private_key.pem。4.2 构建客户端并调用远程智能体现在我们编写客户端代码weather_client.py。假设服务端天气智能体运行在http://localhost:8000并且我们已将其DID配置为信任的调用方在实际场景中服务端可能需要一个授权列表或更复杂的策略。import asyncio import aiohttp from anp.openanp import RemoteAgent from anp.authentication.did_wba import DIDWbaAuth async def main(): # 1. 准备DID认证器 # 加载之前生成的客户端DID文档和私钥 with open(client_did_document.json, r) as f: client_did_document json.load(f) with open(client_private_key.pem, r) as f: client_private_key f.read() # 创建DID-WBA认证器它将自动为请求生成HTTP Message Signatures auth DIDWbaAuth( did_documentclient_did_document, private_key_pemclient_private_key ) # 2. 发现远程智能体 # 提供远程智能体的Agent Description文档URL和认证器 weather_agent_url http://localhost:8000/weather/ad.json print(f Discovering agent at {weather_agent_url}...) try: # RemoteAgent.discover 会自动获取 ad.json 和 interface.json # 并创建一个本地代理对象其方法签名与远程接口一致。 weather_agent await RemoteAgent.discover( agent_description_urlweather_agent_url, authauth # 携带身份进行认证 ) print(f✅ Agent discovered: {weather_agent.config.name}) print(f DID: {weather_agent.config.did}) except aiohttp.ClientError as e: print(f❌ Network error discovering agent: {e}) return except ValueError as e: print(f❌ Failed to parse agent description: {e}) return # 3. 调用远程方法就像调用本地方法一样 print(\n️ Calling remote methods...) try: # 调用 get_current_weather current_weather await weather_agent.get_current_weather(cityBeijing, unitcelsius) print(f✅ Current weather in Beijing: {current_weather}) except Exception as e: # RemoteAgent会将JSON-RPC错误转换为Python异常 print(f❌ Error calling get_current_weather: {e}) try: # 调用 get_forecast forecast await weather_agent.get_forecast(cityShanghai, days2) print(f✅ 2-day forecast for Shanghai: {forecast}) except Exception as e: print(f❌ Error calling get_forecast: {e}) # 4. 调用需要上下文的方法 try: # get_agent_info 方法需要 AgentContext但 RemoteAgent 会自动处理。 # 客户端的DID信息会通过认证头传递服务端可以从context.caller_did获取。 agent_info await weather_agent.get_agent_info() print(f✅ Agent info: {agent_info}) except Exception as e: print(f❌ Error calling get_agent_info: {e}) if __name__ __main__: asyncio.run(main())运行这个客户端你将看到它成功发现了天气智能体并安全地调用了其方法。在服务端的日志中你应该能看到带有Signature等认证头的请求。实操心得错误处理与重试在网络化的智能体调用中错误是常态而非例外。RemoteAgent抛出的异常可能来自网络aiohttp.ClientError、JSON-RPC协议JsonRpcError或远程方法本身ValueError等。在生产代码中你需要根据异常类型实施不同的重试策略。例如网络超时可以重试而参数错误则不应重试。建议使用tenacity等库来实现带退避的智能重试机制。4.3 深入原理RemoteAgent如何工作RemoteAgent.discover()这个魔法方法背后做了很多事情HTTP GET ad.json获取智能体描述文档。解析与验证检查文档格式提取interface.json的URL和rpc端点URL。HTTP GET interface.json获取OpenRPC接口描述。动态创建本地类根据interface.json中的方法定义在运行时动态创建一个Python类。这个类的每个方法都对应远程的一个JSON-RPC调用。封装调用当你调用agent.get_current_weather(...)时动态方法会 a. 使用DIDWbaAuth认证器对请求进行签名。 b. 构造一个符合JSON-RPC 2.0标准的请求体。 c. 发送HTTP POST请求到远程的/rpc端点。 d. 解析响应将JSON-RPC结果转换为Python对象或将错误转换为异常。这一切都对开发者透明你获得的是类型安全通过Pydantic模型和直观的调用体验。5. 进阶实战使用ANP Crawler进行动态发现与LLM集成RemoteAgent适合在代码中明确知道对方接口的场景。但如果你正在构建一个“智能体搜索引擎”或一个由LLM驱动的“智能体调度中枢”你需要更动态的能力。这就是ANPCrawler的用武之地。5.1 Crawler核心工作流解析假设我们想构建一个旅行助手它需要根据用户的自然语言指令如“帮我找一下北京后天天气并推荐一家王府井附近的酒店”动态发现并调用相应的天气和酒店智能体。import asyncio from anp.anp_crawler import ANPCrawler from openai import OpenAI # 假设使用OpenAI API class TravelOrchestrator: def __init__(self, openai_api_key, crawler_did_path, crawler_private_key_path): self.llm_client OpenAI(api_keyopenai_api_key) # 初始化Crawler它需要自己的DID身份去认证 self.crawler ANPCrawler( did_document_pathcrawler_did_path, private_key_pathcrawler_private_key_path ) # 一个简单的智能体注册表实际中可能是一个去中心化的发现服务 self.agent_registry { weather: http://weather-service.example.com/weather/ad.json, hotel: http://hotel-booking.example.com/hotel/ad.json, map: http://maps-service.example.com/poi/ad.json, } async def process_user_query(self, user_query: str): 处理用户查询理解意图 - 发现智能体 - 执行工具 - 整合结果。 print(f Processing query: {user_query}) # 步骤1: 使用LLM分析用户意图并决定需要调用哪些智能体 # 这里简化处理假设LLM直接返回需要调用的智能体类型和参数 # 实际应用中这里会是复杂的提示工程和意图识别 needed_agents [weather, hotel] # 假设LLM分析得出需要天气和酒店 results {} for agent_type in needed_agents: agent_url self.agent_registry.get(agent_type) if not agent_url: print(f⚠️ Agent type {agent_type} not found in registry.) continue print(f️ Crawling agent at {agent_url}) try: # 步骤2: 使用Crawler抓取并解析智能体文档 # fetch_text 会获取原始文档同时将其接口转换为OpenAI Tools格式 raw_content, openai_tools await self.crawler.fetch_text(agent_url) if not openai_tools: print(f⚠️ No tools discovered for {agent_type}.) continue print(f✅ Discovered {len(openai_tools)} tool(s) from {agent_type} agent.) # 步骤3: 将工具描述喂给LLM让LLM决定调用哪个工具以及参数 # 构建一个包含工具定义的LLM对话 system_prompt f你是一个旅行助手。你有以下工具可用 {openai_tools} 请根据用户请求决定是否调用工具并给出正确的参数。 llm_response self.llm_client.chat.completions.create( modelgpt-4, messages[ {role: system, content: system_prompt}, {role: user, content: user_query} ], toolsopenai_tools, # 关键将ANP接口作为工具提供给LLM tool_choiceauto ) # 步骤4: 处理LLM的响应可能包含工具调用 message llm_response.choices[0].message if message.tool_calls: for tool_call in message.tool_calls: tool_name tool_call.function.name tool_args json.loads(tool_call.function.arguments) print(f LLM decided to call tool: {tool_name} with args {tool_args}) # 步骤5: 使用Crawler执行工具调用 # execute_tool_call 内部会找到对应方法构造JSON-RPC请求并发送 tool_result await self.crawler.execute_tool_call( tool_nametool_name, argumentstool_args, # Crawler知道从哪里发送请求因为它之前fetch过agent_url ) results[f{agent_type}_{tool_name}] tool_result print(f✅ Tool call result: {tool_result}) else: print(fℹ️ LLM did not call any tools for {agent_type}.) except Exception as e: print(f❌ Error interacting with {agent_type} agent: {e}) continue # 步骤6: 整合所有结果生成最终回复 final_answer self._synthesize_results(results, user_query) return final_answer def _synthesize_results(self, results: dict, original_query: str) - str: # 这里可以再次调用LLM将分散的结果整合成一段连贯的回复 # 简化处理直接返回结果摘要 summary f根据您的查询『{original_query}』已获取以下信息\n for key, value in results.items(): summary f- {key}: {value}\n return summary # 使用示例 async def main(): orchestrator TravelOrchestrator( openai_api_keyyour-api-key, crawler_did_pathpath/to/orchestrator_did.json, crawler_private_key_pathpath/to/orchestrator_private.pem ) answer await orchestrator.process_user_query( 北京后天天气怎么样再推荐一家王府井附近的五星级酒店。 ) print(\n *50) print( Final Answer:) print(answer) if __name__ __main__: asyncio.run(main())这个例子展示了ANPCrawler的核心价值它充当了LLM与ANP智能体网络之间的通用适配器。fetch_text方法将ANP的interface.json转换成了LLM能理解的OpenAI Tools格式而execute_tool_call则负责将LLM的工具调用决策翻译成具体的JSON-RPC请求。5.2 Crawler与RemoteAgent的深度对比为了更清晰地选择我们深入对比两者特性维度RemoteAgentANP Crawler设计哲学面向对象代理。为每个远程智能体创建一个本地的、类型化的代理对象。面向过程爬虫。提供一组函数来获取文档、解析工具、执行调用。类型安全高。方法签名在发现时动态生成IDE可以进行代码补全和类型检查。低。工具名称和参数以字典形式传递类型检查在运行时进行。代码风格result await agent.search(queryTokyo)如同调用本地库。result await crawler.execute_tool_call(search, {query: Tokyo})如同调用一个通用API。适用场景1.服务提供方构建稳定、类型清晰的智能体服务。2.强类型客户端在代码中明确知道并依赖特定智能体接口的应用。1.动态调度器/中介需要根据运行时信息动态发现和调用未知智能体。2.LLM集成将任意ANP智能体的能力作为“工具”暴露给LLM。3.监控/探索工具编写扫描整个智能体网络的工具。性能考量首次发现discover有HTTP开销后续调用无额外发现开销。每次fetch_text都可能产生HTTP开销。可通过缓存interface.json来优化。错误处理异常与本地Python异常集成度高易于用try...except捕获特定错误。错误通常以字典或通用异常形式返回需要手动解析错误码和消息。决策指南如果你的应用架构是“一个中心智能体调用N个已知的、稳定的伙伴智能体”用RemoteAgent开发体验更好。如果你的应用是“一个中介平台需要对接海量不确定的、动态注册的智能体”或者核心是“LLM驱动的工作流”那么ANPCrawler的灵活性是不可替代的。6. 生产环境部署与故障排查实录将基于AgentConnect的智能体部署到生产环境会面临与普通Web服务不同的一系列挑战。以下是我在实际部署中积累的经验和常见问题的解决方案。6.1 安全配置清单安全是智能体网络的命脉。以下清单必须逐项核对DID私钥管理禁止将私钥文件放在代码仓库、镜像或环境变量中环境变量仍有泄露风险。推荐使用云服务商的密钥管理服务KMS。修改认证代码从KMS API动态获取私钥进行签名。# 伪代码示例从AWS Secrets Manager获取私钥 import boto3 from botocore.exceptions import ClientError def get_private_key_from_aws(): secret_name anp-agent-private-key region_name us-east-1 session boto3.session.Session() client session.client(service_namesecretsmanager, region_nameregion_name) try: response client.get_secret_value(SecretIdsecret_name) return response[SecretString] # 你的PEM格式私钥 except ClientError as e: # 处理错误 raise服务端身份验证与授权基础验证在AgentConfig中设置auth_requiredTrue确保所有请求都经过DID签名验证。访问控制列表ACL实现一个简单的ACL只允许特定的DID或DID模式调用你的智能体。可以在AgentContext中获取caller_did并进行校验。from fastapi import HTTPException ALLOWED_DIDS {did:wba:trusted-client.com:agent, did:wba:partner-org.net:*} anp_agent(AgentConfig(..., auth_requiredTrue)) class MySecureAgent: interface async def sensitive_operation(self, context: AgentContext, data: str): caller_did context.caller_did if not any(caller_did allowed or allowed.endswith(*) and caller_did.startswith(allowed[:-1]) for allowed in ALLOWED_DIDS): raise HTTPException(status_code403, detailCaller DID not authorized.) # ... 业务逻辑网络与传输安全强制HTTPS生产环境必须使用HTTPS。ad.json中的serviceEndpoint和所有链接都应使用https://。CORS配置如果你的智能体需要被浏览器端的应用调用需在FastAPI中正确配置CORS。但注意ANP主要面向服务间通信CORS非必需。速率限制使用像slowapi或fastapi-limiter这样的中间件防止恶意调用耗尽资源。6.2 性能优化与可观测性文档缓存RemoteAgent.discover()和ANPCrawler.fetch_text()都会下载JSON文档。对于不常变化的智能体应该缓存这些文档。可以为RemoteAgent实现一个带TTL的缓存装饰器。from functools import lru_cache from datetime import datetime, timedelta class CachedRemoteAgent: def __init__(self): self._cache {} async def get_agent(self, url, auth): cache_key (url, auth.did) # 根据URL和认证者DID缓存 if cache_key in self._cache: entry self._cache[cache_key] if datetime.now() entry[expiry]: return entry[agent] # 缓存未命中或过期执行发现 agent await RemoteAgent.discover(url, auth) self._cache[cache_key] { agent: agent, expiry: datetime.now() timedelta(minutes30) # 缓存30分钟 } return agent日志与监控结构化日志使用structlog或json-logging记录所有RPC调用包括调用者DID、方法名、耗时、成功/失败状态。这对于审计和调试至关重要。指标暴露使用prometheus-client在/metrics端点暴露指标如请求总数、按方法分类的请求数、错误数、请求延迟分布等。便于使用Grafana进行监控。分布式追踪在微服务架构中为每个ANP请求注入唯一的追踪ID如X-Trace-Id并贯穿整个调用链便于排查跨智能体的性能问题。6.3 常见问题排查速查表在实际运维中你会频繁遇到以下问题。这里提供一个快速排查指南。问题现象可能原因排查步骤与解决方案客户端discover()失败报ValueError或JSONDecodeError1. 网络不通或URL错误。2. 服务端未返回有效的ad.json。3. 返回的JSON格式不符合ANP规范。1. 用curl或浏览器直接访问ad.json的URL确认可访问且返回JSON。2. 检查服务端AgentConfig中的prefix是否正确确保/ad.json端点已正确注册。3. 验证返回的JSON是否包含必需的name,did,interfaces等字段。客户端RPC调用返回401 Unauthorized或403 Forbidden1. 客户端未提供认证信息。2. 客户端DID私钥错误或与文档不匹配。3. 服务端启用了ACL且当前DID不在允许列表中。4. 时钟不同步HTTP签名包含时间戳。1. 确认创建DIDWbaAuth时传入了正确的DID文档和私钥。2. 使用工具重新生成密钥对和DID文档确保匹配。3. 检查服务端日志查看ACL拒绝的具体原因。4. 确保服务器和客户端时间与NTP服务器同步误差在允许范围内通常±60秒。客户端调用方法时抛出JsonRpcError错误码-32601(Method not found)1. 客户端调用的方法名与interface.json中定义的不符。2. 服务端接口已更新客户端缓存的interface.json过期。1. 直接访问服务端的/interface.json确认方法名和参数列表。2. 清除客户端缓存或实现缓存失效逻辑强制重新发现。服务端收到请求但AgentContext.caller_did为None1. 请求未携带有效的DID认证头。2. 请求签名验证失败。3.auth_required设置为False但代码又尝试读取caller_did。1. 检查客户端是否使用了DIDWbaAuth。2. 检查服务端日志中的签名验证错误信息。3. 如果方法需要调用者身份确保AgentConfig(auth_requiredTrue)并在方法参数中声明context: AgentContext。服务端高并发下性能下降或超时1. 智能体业务逻辑本身是CPU或IO密集型。2. 每个请求都重复进行复杂的DID解析和签名验证。1. 对智能体方法进行性能剖析优化慢查询或算法。2. 考虑对已验证的DID签名进行短期缓存如5分钟避免重复的密码学运算。使用lru_cache装饰验证函数。LLM CrawlerLLM无法正确调用工具1.fetch_text转换的OpenAI Tools格式描述不清。2. 工具参数过于复杂LLM无法理解。3. 网络超时导致工具调用失败LLM收不到结果。1. 在智能体方法的docstring中提供清晰、结构化的描述包括参数示例。2. 简化接口设计避免嵌套过深的复杂对象。3. 在execute_tool_call外层添加重试和超时控制并为LLM提供明确的错误反馈例如“调用酒店服务超时请稍后再试”。6.4 调试技巧深入请求内部当问题复杂时需要查看原始的HTTP请求和响应。为aiohttp会话启用调试日志非常有用。import logging import sys import asyncio from anp.openanp import RemoteAgent from anp.authentication.did_wba import DIDWbaAuth # 启用aiohttp的调试日志 logging.basicConfig(levellogging.DEBUG) loggers [aiohttp.client, aiohttp.server, aiohttp.internal] for logger_name in loggers: logger logging.getLogger(logger_name) logger.setLevel(logging.DEBUG) logger.addHandler(logging.StreamHandler(sys.stdout)) async def debug_call(): auth DIDWbaAuth(...) # 你的认证信息 agent await RemoteAgent.discover(http://localhost:8000/weather/ad.json, auth) # 这次调用会在控制台打印出所有HTTP请求和响应的头信息及正文部分 result await agent.get_current_weather(cityBeijing) print(result) asyncio.run(debug_call())通过日志你可以清晰地看到发出的HTTP请求是否包含了正确的Signature-Input和Signature头以及服务器返回的具体错误信息这是定位认证和网络问题最直接的方法。构建基于ANP的智能体网络就像在铺设数字世界的新一代“铁路网”。AgentConnect SDK提供了铁轨、枕木和信号系统协议与安全而你的业务逻辑则是奔跑其上的列车。从定义一个简单的interface装饰的方法开始你就已经迈入了这个可编程、可组合的智能体未来。记住从简单的原型出发逐步迭代重视安全和可观测性你的智能体就能在日益壮大的网络中可靠地运行并创造价值。