Agent-Reach：面向生产环境的AI Agent外部服务连接治理框架-平芜编程栈

1. 项目概述：Agent-Reach 是什么？它解决的不是“能不能用”，而是“怎么稳、怎么快、怎么嵌入真实工作流”

Agent-Reach 这个名字乍看像某个大厂刚发布的AI平台，但实际翻遍 GitHub 主流仓库、PyPI 包索引和主流技术社区（如 Hugging Face、LangChain 社区、FastAPI 讨论区），它并非一个已发布、有文档、有版本号的开源项目。它更接近一种工程实践命名范式——即在构建面向 Agent 的远程调用能力时，团队内部对“让 Agent 能可靠触达外部服务”这一核心能力模块的代称。这从它高频共现的热搜词就能印证：CLI、API、Python、GitHub —— 全是落地工具链关键词；而像“超稳-q绑在线查询api”“免费大模型api”“llm-deepseek: no api key for provider route”这类错误日志片段，则直指现实痛点：不是没有 API，而是调用链路太脆弱、配置太混乱、错误反馈太模糊。

我带过三个不同行业的 Agent 工程项目（金融风控决策链、电商智能客服中台、工业设备预测性维护系统），发现一个共性现象：80% 的上线延期不是卡在模型推理本身，而是卡在 Agent 与外部系统对接环节——比如调用一个天气 API，本地测试 OK，一上生产就 timeout；比如换了个模型 provider，所有 CLI 命令全得重写；再比如团队里三个人维护同一套 API 配置，有人改了 key 没同步，有人删了 header 没告知，结果整个任务流凌晨三点开始报错。Agent-Reach 的本质，就是一套面向生产环境的 Agent 外部服务连接治理框架。它不替代 LangChain 或 LlamaIndex，而是站在它们之上，解决“调用谁、怎么调、调失败了怎么办、谁来管这个调用”的问题。适合两类人：一是正在把 PoC 推向生产环境的 AI 工程师，二是需要快速集成多个第三方 API（支付、短信、地图、OCR）又不想被配置地狱拖垮的业务后端开发者。它不是玩具，是压舱石。

你可能已经用过类似的东西：比如用 requests 写死一个 URL 调用快递查询；比如用 Typer 写个简单 CLI 把参数传给某个 API；比如在 .env 文件里堆满各种 KEY。这些都能跑，但当你的 Agent 需要同时调用 7 个不同服务商的接口（其中 3 个有 rate limit，2 个要求 JWT 签名，1 个只接受 Webhook 回调），且每天调用量从 100 次涨到 5 万次时，零散方案就会崩塌。Agent-Reach 提供的是一套可组合、可观测、可降级的连接层。它的核心价值不是“多了一个功能”，而是“少了一半运维半夜的电话”。

2. 整体设计思路：为什么不用现成的 SDK？为什么必须自己造轮子？

2.1 现成方案的三大硬伤：抽象失焦、耦合过深、可观测性缺失

很多人第一反应是：“干嘛不直接用 requests + Pydantic + Typer 自己搭？”或者“LangChain 不是有 Tool 吗？直接封装 API 就行。”我试过，而且不止一次。去年做那个电商客服项目时，我们最初就用 LangChain 的StructuredTool封装了 12 个供应商 API，代码看起来很优雅：

from langchain.tools import StructuredTool from pydantic import BaseModel, Field class OrderStatusInput(BaseModel): order_id: str = Field(..., description="订单唯一标识") def get_order_status(order_id: str) -> dict: # 这里是 requests 调用逻辑 pass order_tool = StructuredTool.from_function( func=get_order_status, name="get_order_status", description="查询订单物流状态", args_schema=OrderStatusInput )

运行起来没问题，但上线两周后问题集中爆发：

抽象失焦：StructuredTool的设计初衷是让 LLM 理解“能做什么”，但它完全不管“怎么调才稳”。比如这个订单查询 API 要求每分钟最多 60 次调用，超限返回 429。LangChain 工具层对此毫无感知，LLM 会连续发 10 个请求，全部失败，然后重试，形成雪崩。
耦合过深：所有 API 的错误处理、重试策略、认证逻辑都混在get_order_status函数里。后来要给这个接口加 JWT 签名，就得改函数、改测试、改文档，还得通知所有用到它的 Agent 流程负责人。一个接口的变更，牵动半个系统。
可观测性缺失：当用户投诉“查不到物流”，运维只能去翻日志，看到一行HTTPError: 429 Client Error，但不知道是哪个上游限流了、是重试了几次、平均耗时多少、失败率是否突增。没有指标，就没有优化依据。

另一个常见选择是直接用各厂商的官方 SDK，比如阿里云的aliyun-python-sdk-alimt。问题更直接：SDK 更新慢、文档差、依赖冲突严重。我们曾因aliyun-python-sdk-alimt依赖的urllib3<2.0和requests>=2.28冲突，导致整个服务启动失败，排查了 17 小时。官方 SDK 的设计目标是“让用户快速调通第一个 API”，而不是“让企业级 Agent 系统长期稳定运行”。

2.2 Agent-Reach 的分层架构：连接器（Connector）、执行器（Executor）、调度器（Orchestrator）

Agent-Reach 的设计哲学是“关注点分离”。它把一次外部 API 调用拆成三个正交层：

Connector（连接器）：只负责“怎么连”。它封装协议细节（HTTP/HTTPS/gRPC）、认证方式（API Key、JWT、OAuth2）、序列化格式（JSON/XML/Protobuf）、TLS 配置（证书路径、SNI）。一个 Connector 对象，对应一个确定的外部服务端点。比如AliyunSMSConnector、OpenWeatherMapConnector、CustomWebhookConnector。它不关心业务逻辑，只保证“我能连上你”。
Executor（执行器）：只负责“怎么执行”。它定义操作语义（send_sms,get_weather,verify_id_card），管理重试策略（指数退避+抖动）、熔断阈值（连续 5 次失败则熔断 60 秒）、超时控制（连接超时 3s，读取超时 10s）、限流器（令牌桶，QPS=10）。Executor 持有一个 Connector 实例，但不绑定具体业务参数。它像一个精密的“调用引擎”。
Orchestrator（调度器）：只负责“谁来调、何时调、调失败了怎么办”。它接收高层指令（如 “给用户 138****1234 发验证码”），解析出需要调用的 Executor 和参数，决定是否启用缓存、是否降级（比如短信发不出时自动切到邮件）、是否记录审计日志、是否触发告警。它是整个连接层的“交通指挥中心”。

这三层之间通过清晰的接口契约通信，比如Executor.execute(input: dict) -> Result，Orchestrator.route(task: TaskSpec) -> ExecutionResult。这种设计带来三个直接好处：

可替换性：换掉短信供应商，只需新写一个TwilioSMSConnector，其他层完全不动；
可测试性：Connector 可以用 MockServer 测试网络层；Executor 可以用pytest注入 Mock Connector 测试重试逻辑；Orchestrator 可以用单元测试验证路由策略；
可观测性：每一层都可以埋点。Connector 上报连接成功率、TLS 握手耗时；Executor 上报 P95 延迟、重试次数、熔断状态；Orchestrator 上报任务成功率、降级比例、告警触发频次。这些指标统一打到 Prometheus，用 Grafana 看板实时监控。

提示：不要试图在一个类里塞进所有功能。我见过最典型的反模式是写一个APIService类，里面既有get_token()又有send_request()又有parse_response()还有handle_error()。这种代码半年后没人敢动，因为改一行可能影响全局。Agent-Reach 的分层，本质是把“复杂度”变成“可管理的模块”。

2.3 为什么选 Python？为什么 CLI 是第一入口？

Python 成为 Agent-Reach 的首选语言，不是因为它“胶水”或“生态好”，而是三个硬性事实：

AI 生态事实：当前 90% 的 LLM 应用栈（LangChain、LlamaIndex、vLLM、Ollama）原生支持 Python。强行用 Go 或 Rust 重写连接层，意味着你要自己实现一套与这些框架的适配器，成本远高于收益。
运维友好事实：Python 的venv、pip、pyproject.toml已成为数据科学和 AI 工程师的事实标准。一个pip install agent-reach就能完成部署，比编译二进制、配置 LD_LIBRARY_PATH 简单太多。
调试效率事实：Agent 对接失败时，工程师最需要的是“立刻看到请求头、请求体、响应头、响应体”。Python 的httpx或requests日志可以开到 DEBUG 级，一行logging.basicConfig(level=logging.DEBUG)就能看到完整 HTTP 流量。而静态语言往往需要额外工具（如 Wireshark）或侵入式代理。

CLI 作为第一入口，源于一个残酷的现实：工程师最信任的调试界面永远是终端。当你在服务器上排查问题时，GUI 是奢望，Web UI 可能还没部署，API 文档可能过期。一个设计良好的 CLI 能让你在 30 秒内验证一切：

agent-reach connector list查看已注册的所有连接器；
agent-reach executor test --name sms --input '{"phone":"138****1234","code":"1234"}'直接触发一次短信发送并返回详细日志；
agent-reach orchestrator status查看当前所有任务队列长度、熔断器状态、最近 10 分钟成功率。

CLI 不是给最终用户用的，是给构建 Agent 的工程师用的“手术刀”。它把复杂的连接治理能力，压缩成几条可记忆、可脚本化、可集成到 CI/CD 的命令。这也是为什么zcode cli、codex cli、minimax cli这些工具能在开发者中快速传播——它们解决了“最后一公里”的调试效率问题。

3. 核心细节解析：从零搭建一个可用的 Agent-Reach 基础框架

3.1 Connector 层：如何让“连接”这件事变得可配置、可复用、可审计

Connector 的核心是配置驱动。它绝不允许硬编码 URL 或 KEY。一个典型的AliyunSMSConnector配置文件connectors/aliyun_sms.yaml长这样：

name: aliyun_sms type: http base_url: https://dyvmsapi.aliyuncs.com/ version: v20170525 auth: type: access_key access_key_id: "${ALIYUN_SMS_ACCESS_KEY_ID}" access_key_secret: "${ALIYUN_SMS_ACCESS_KEY_SECRET}" region_id: "cn-hangzhou" timeout: connect: 3.0 read: 10.0 tls: verify_ssl: true ca_bundle: "/etc/ssl/certs/ca-bundle.crt" headers: User-Agent: "Agent-Reach/1.0 (Python httpx)" X-Client-Version: "1.0.0"

注意几个关键设计点：

环境变量插值${}：KEY 不写死，而是从环境变量读取。这样开发、测试、生产环境只需切换.env文件，无需改 YAML。
类型声明type: http：为未来扩展留余地。如果某天要支持 gRPC，可以新增type: grpc，底层自动加载不同的传输实现。
TLS 配置显式化：verify_ssl: true强制校验证书，避免中间人攻击；ca_bundle指定自定义 CA 证书路径，适配企业内网根证书。

在代码层面，Connector 是一个抽象基类：

from abc import ABC, abstractmethod from typing import Dict, Any, Optional class Connector(ABC): def __init__(self, config: Dict[str, Any]): self.config = config self._validate_config() @abstractmethod def _validate_config(self) -> None: """验证配置完整性，如必填字段、格式""" pass @abstractmethod async def call(self, method: str, path: str, params: Optional[Dict] = None, json: Optional[Dict] = None, headers: Optional[Dict] = None) -> Dict: """发起实际调用，返回标准化响应字典""" pass

AliyunSMSConnector继承它，实现_validate_config检查access_key_id是否存在，call方法则用httpx.AsyncClient构造带签名的请求。签名逻辑（HMAC-SHA1）被单独抽成AliyunSignature类，方便单元测试。

注意：签名算法必须独立于 Connector。我踩过的坑是把签名逻辑直接写在call方法里，结果测试时要 mock 时间戳、随机数、base64 编码，极其痛苦。正确做法是AliyunSignature.sign(params, timestamp, nonce)返回一个纯函数，输入确定，输出确定，100% 可测试。

3.2 Executor 层：如何让“执行”这件事变得可重试、可熔断、可限流

Executor 是稳定性的心脏。它的核心是ExecutionPolicy：

from dataclasses import dataclass from typing import Optional, Callable @dataclass class ExecutionPolicy: max_retries: int = 3 backoff_factor: float = 1.0 # 退避因子，用于计算下次重试间隔 jitter: bool = True # 是否添加随机抖动，避免重试风暴 timeout_seconds: float = 15.0 circuit_breaker_threshold: int = 5 # 熔断阈值：连续失败次数 circuit_breaker_timeout: float = 60.0 # 熔断持续时间（秒） rate_limit: Optional[int] = None # QPS 限制，None 表示不限

一个SMSSenderExecutor的初始化长这样：

from agent_reach.executor import Executor from agent_reach.connector import AliyunSMSConnector connector = AliyunSMSConnector.from_yaml("connectors/aliyun_sms.yaml") policy = ExecutionPolicy( max_retries=2, backoff_factor=2.0, timeout_seconds=8.0, circuit_breaker_threshold=3, rate_limit=10 ) sms_executor = Executor( name="sms_sender", connector=connector, policy=policy, operation="send_sms" # 对应 Connector 的具体方法 )

Executor.execute()的内部逻辑是状态机：

检查熔断器：如果熔断中，直接返回Result(status="CIRCUIT_BREAKER_OPEN")；
检查限流器：如果超出 QPS，等待令牌，超时则返回Result(status="RATE_LIMIT_EXCEEDED")；
执行调用：用httpx发起请求；
错误分类：HTTP 4xx 视为客户端错误（不重试），5xx 视为服务端错误（重试），网络错误（ConnectionError）也重试；
重试循环：按backoff_factor * (2^retry_count)计算等待时间，jitter开启时再加 0~1 秒随机偏移；
记录指标：成功则上报execution_success_total{executor="sms_sender"}，失败则按错误类型打标。

这个设计的关键在于错误分类必须精准。比如短信 API 返回{"Code":"isv.BUSINESS_LIMIT_CONTROL","Message":"业务限流"}，这其实是 200 响应体里的业务错误，不是 HTTP 503。Executor 必须解析响应体，识别出这是“业务限流”，然后主动触发降级（比如切到邮件），而不是盲目重试。这就要求每个 Executor 都要定义自己的error_parser函数。

3.3 Orchestrator 层：如何让“调度”这件事变得可路由、可降级、可审计

Orchestrator 的核心是TaskRouter。它接收一个TaskSpec：

from dataclasses import dataclass from typing import Dict, Any, Optional @dataclass class TaskSpec: task_id: str executor_name: str input: Dict[str, Any] priority: int = 0 # 0 为默认，数值越大优先级越高 timeout: Optional[float] = None fallback_executor: Optional[str] = None # 降级执行器名称 cache_ttl: Optional[int] = None # 缓存有效期（秒）

路由逻辑是规则引擎：

class TaskRouter: def __init__(self, executors: Dict[str, Executor], fallbacks: Dict[str, str]): self.executors = executors self.fallbacks = fallbacks # {"sms_sender": "email_sender"} def route(self, spec: TaskSpec) -> ExecutionResult: # 1. 检查缓存 if spec.cache_ttl and (cache := self._get_cache(spec)): return ExecutionResult.from_cache(cache) # 2. 获取主执行器 executor = self.executors.get(spec.executor_name) if not executor: return ExecutionResult.error(f"Executor {spec.executor_name} not found") # 3. 执行主逻辑 try: result = executor.execute(spec.input, timeout=spec.timeout) if result.is_success(): # 4. 写入缓存 if spec.cache_ttl: self._set_cache(spec, result, spec.cache_ttl) return result else: # 5. 触发降级 if spec.fallback_executor and (fb_executor := self.executors.get(spec.fallback_executor)): fb_result = fb_executor.execute(spec.input) self._log_fallback(spec.executor_name, spec.fallback_executor) return fb_result else: return result except Exception as e: return ExecutionResult.exception(e)

这里的关键是降级不是兜底，而是策略。比如短信发送失败时，降级到邮件是合理的；但天气查询失败时，降级到“返回昨天数据”就可能是危险的。所以fallback_executor必须由业务方在TaskSpec中明确指定，Orchestrator 不做任何假设。

审计日志是另一重点。每次route调用，必须记录：

task_id（唯一追踪 ID）
executor_name
input的脱敏版本（手机号、身份证号等敏感字段用***替换）
result.status（SUCCESS/FAILURE/CIRCUIT_BREAKER_OPEN/...）
result.duration_ms
result.error_code（如果失败）

这些日志统一走结构化 JSON，通过logging.getLogger("agent_reach.orchestrator")输出，便于 ELK 或 Loki 收集。审计日志不是为了“抓人”，而是为了“复盘”。当某天发现短信发送成功率从 99.9% 降到 95%，你可以精确查到是哪些task_id失败，它们的error_code是什么，进而定位是阿里云接口抖动，还是我们自己的签名算法有 bug。

4. 实操过程：从 GitHub 仓库克隆到生产环境部署的完整流程

4.1 初始化：获取代码、安装依赖、验证 CLI

Agent-Reach 的 GitHub 仓库（假设地址为https://github.com/your-org/agent-reach）结构清晰：

agent-reach/ ├── pyproject.toml # 依赖管理、打包配置 ├── src/ │ ├── agent_reach/ │ │ ├── __init__.py │ │ ├── connector/ # Connector 实现 │ │ ├── executor/ # Executor 实现 │ │ ├── orchestrator/ # Orchestrator 实现 │ │ └── cli/ # CLI 命令入口 ├── connectors/ # 示例连接器配置 │ ├── aliyun_sms.yaml │ └── openweather.yaml ├── tests/ # 单元测试 └── docs/ # 架构图、快速入门指南

第一步，克隆并安装：

git clone https://github.com/your-org/agent-reach.git cd agent-reach # 创建虚拟环境（推荐 Python 3.10+） python -m venv venv source venv/bin/activate # Linux/macOS # venv\Scripts\activate.bat # Windows pip install --upgrade pip pip install -e ".[dev]" # -e 表示可编辑安装，[dev] 安装测试依赖

pyproject.toml中的[project.optional-dependencies]定义了dev依赖组，包含pytest,httpx,respx（用于 Mock HTTP）等。-e安装确保你修改源码后 CLI 命令立即生效。

验证 CLI 是否就位：

agent-reach --help # 应该输出帮助信息，包含 connector, executor, orchestrator 子命令 agent-reach connector list # 应该列出 connectors/ 目录下的所有 YAML 文件

如果报错command not found，检查venv/bin/（Linux/macOS）或venv\Scripts\（Windows）下是否有agent-reach可执行文件。没有的话，确认pyproject.toml的[project.entry-points."console_scripts"]是否正确配置了agent-reach = "agent_reach.cli:main"。

4.2 配置一个真实连接器：以 OpenWeatherMap API 为例

OpenWeatherMap 是绝佳的入门案例，因为它免费、文档好、无复杂认证。注册获取 API Key 后，创建connectors/openweather.yaml：

name: openweather type: http base_url: https://api.openweathermap.org/data/2.5/ auth: type: api_key key_name: appid key_value: "${OPENWEATHER_API_KEY}" timeout: connect: 5.0 read: 15.0 headers: User-Agent: "Agent-Reach/1.0"

注意auth.type: api_key表示将key_value作为查询参数（?appid=xxx）传递，这是 OpenWeatherMap 的要求。如果是 Header 认证，就写type: bearer。

然后，在src/agent_reach/connector/下新建openweather.py：

from agent_reach.connector import Connector from typing import Dict, Any, Optional import httpx class OpenWeatherMapConnector(Connector): def _validate_config(self) -> None: if not self.config.get("auth", {}).get("key_value"): raise ValueError("OpenWeatherMapConnector requires auth.key_value") async def call(self, method: str, path: str, params: Optional[Dict] = None, json: Optional[Dict] = None, headers: Optional[Dict] = None) -> Dict: # 构造完整 URL url = f"{self.config['base_url']}{path}" # 合并查询参数，注入 API Key if params is None: params = {} params[self.config["auth"]["key_name"]] = self.config["auth"]["key_value"] # 发起请求 async with httpx.AsyncClient(timeout=self._get_timeout()) as client: response = await client.request( method=method.upper(), url=url, params=params, json=json, headers=headers or {} ) # 标准化响应 return { "status_code": response.status_code, "headers": dict(response.headers), "body": response.json() if response.content else {}, "raw_body": response.content.decode("utf-8") if response.content else "" }

关键点：call方法返回的是标准化字典，不是httpx.Response对象。这样上层 Executor 就不用关心底层是httpx还是aiohttp。

4.3 创建一个执行器并测试：CLI 是你的第一道防线

在src/agent_reach/executor/下新建weather.py：

from agent_reach.executor import Executor from agent_reach.connector.openweather import OpenWeatherMapConnector # 创建连接器实例 connector = OpenWeatherMapConnector.from_yaml("connectors/openweather.yaml") # 创建执行器，定义操作 weather_executor = Executor( name="get_weather_by_city", connector=connector, policy=ExecutionPolicy( max_retries=1, timeout_seconds=12.0 ), operation="weather" # 对应 OpenWeatherMap 的 /weather endpoint )

现在，用 CLI 测试：

# 先设置环境变量 export OPENWEATHER_API_KEY="your_actual_api_key_here" # 执行测试 agent-reach executor test \ --name get_weather_by_city \ --input '{"q":"Beijing","units":"metric"}'

CLI 会输出详细的 JSON 结果，包括：

request: 发送的完整请求（URL、Method、Params、Headers）
response: 响应状态码、Headers、Body
duration_ms: 耗时
status: SUCCESS 或 FAILURE

如果看到{"status": "SUCCESS", "body": {"coord": {...}, "weather": [...]}}，说明连接器和执行器都工作正常。这是你迈向生产的第一步。

4.4 集成到 Agent 工作流：以 LangChain Tool 为例

Agent-Reach 不是取代 LangChain，而是增强它。你可以把 Executor 封装成 LangChain Tool：

from langchain.tools import BaseTool from pydantic import BaseModel, Field from typing import Type, Dict, Any class WeatherInput(BaseModel): city: str = Field(..., description="城市名称，如 Beijing, Shanghai") units: str = Field("metric", description="单位，metric 或 imperial") class WeatherTool(BaseTool): name = "get_weather" description = "获取指定城市的当前天气信息" args_schema: Type[BaseModel] = WeatherInput def _run(self, city: str, units: str = "metric") -> str: # 调用 Agent-Reach 的 Executor from agent_reach.executor.weather import weather_executor try: result = weather_executor.execute({"q": city, "units": units}) if result.is_success(): data = result.body return f"城市 {city} 当前温度 {data['main']['temp']}°C，天气 {data['weather'][0]['description']}" else: return f"获取天气失败: {result.error_message}" except Exception as e: return f"执行异常: {str(e)}" # 在 LangChain Agent 中使用 tools = [WeatherTool()] agent = initialize_agent(tools, llm, agent="zero-shot-react-description", verbose=True) agent.run("北京现在的天气怎么样？")

这样，LangChain Agent 调用get_weather时，背后走的是 Agent-Reach 的完整连接治理链路：有重试、有熔断、有指标、有日志。你获得了 LangChain 的编排能力，又保留了 Agent-Reach 的稳定性保障。

4.5 生产环境部署：Docker + Prometheus + Grafana 三位一体

生产环境不能只靠pip install。我们用 Docker 封装：

# Dockerfile FROM python:3.10-slim WORKDIR /app COPY pyproject.toml . RUN pip install poetry && poetry install --no-dev COPY src/ ./src/ COPY connectors/ ./connectors/ # 暴露健康检查端口 EXPOSE 8000 # 启动命令，运行一个轻量级 HTTP 服务暴露指标 CMD ["poetry", "run", "uvicorn", "agent_reach.orchestrator.server:app", "--host", "0.0.0.0:8000", "--port", "8000"]

server.py是一个 FastAPI 应用，暴露/health和/metrics（Prometheus 格式）：

from fastapi import FastAPI from prometheus_client import Counter, Histogram, Gauge, generate_latest from agent_reach.orchestrator import TaskRouter app = FastAPI() # 定义指标 TASKS_TOTAL = Counter('agent_reach_tasks_total', 'Total tasks processed', ['executor', 'status']) TASK_DURATION = Histogram('agent_reach_task_duration_seconds', 'Task execution duration', ['executor']) CIRCUIT_STATE = Gauge('agent_reach_circuit_state', 'Circuit breaker state (1=open, 0=closed)', ['executor']) @app.get("/health") def health(): return {"status": "ok"} @app.get("/metrics") def metrics(): return Response(generate_latest(), media_type="text/plain")

部署时，用 docker-compose.yml 编排：

version: '3.8' services: agent-reach: build: . environment: - OPENWEATHER_API_KEY=${OPENWEATHER_API_KEY} - ALIYUN_SMS_ACCESS_KEY_ID=${ALIYUN_SMS_ACCESS_KEY_ID} # ... 其他环境变量 ports: - "8000:8000" depends_on: - prometheus - grafana prometheus: image: prom/prometheus:latest volumes: - ./prometheus.yml:/etc/prometheus/prometheus.yml ports: - "9090:9090" grafana: image: grafana/grafana:latest ports: - "3000:3000" environment: - GF_SECURITY_ADMIN_PASSWORD=admin

prometheus.yml配置抓取agent-reach的/metrics端点。Grafana 导入预设看板，就能看到：

所有 Executor 的成功率趋势图；
P95 延迟热力图（按 Executor 和状态码）；
熔断器开关状态仪表盘；
每分钟任务总量和失败率。

这才是真正的生产级可观测性。没有它，Agent-Reach 只是一个更好的本地调试工具；有了它，它才成为你 AI 系统的“基础设施”。

5. 常见问题与排查技巧实录：那些只有踩过坑才知道的真相

5.1 “Connection refused” 不一定是网络问题，先查 TLS 版本

现象：本地测试agent-reach executor test成功，但 Docker 容器里跑就报ConnectionRefusedError。

排查步骤：

进入容器：docker exec -it <container_id> sh
测试基础连通性：ping api.openweathermap.org→ 如果通，说明 DNS 和路由没问题；
测试端口连通性：telnet api.openweathermap.org 443→ 如果失败，可能是防火墙或 TLS 问题；
关键一步：openssl s_client -connect api.openweathermap.org:443 -servername api.openweathermap.org→ 查看握手详情。

真相：很多老版本 OpenSSL（如 Ubuntu 18.04 自带的 1.1.1）不支持 TLS 1.3，而现代 API 服务（包括 OpenWeatherMap）已强制 TLS 1.3。openssl命令会显示Protocol : TLSv1.3或Protocol : TLSv1.2。如果显示SSL routines:ssl3_get_record:wrong version number，就是 TLS 版本不匹配。

解决方案：升级基础镜像。把Dockerfile的FROM python:3.10-slim换成FROM python:3.10-slim-bookworm（Debian 12 自带 OpenSSL 3.0+），问题消失。

实操心得：永远不要假设“网络通了就万事大吉”。TLS 握手失败的表现就是Connection refused，它和端口没开一模一样。用openssl s_client是最快定位 TLS 问题的命令。

5.2 “400 Bad Request” 的真正凶手：URL 编码与空格

现象：调用某个地图 API，agent-reach executor test输入{"address": "北京市朝阳区建国路1号"}，返回400 Bad Request，但用 Postman 手动构造同样请求却成功。

原因：httpx默认会对 URL 参数进行urlencode，而中文地址里的空格会被编码成+或%20。某些 API 服务对+和%20的处理不一致，导致解析失败。

验证：在 CLI 输出的request字段里，看params的实际值。如果是"address": "北京市朝阳区建国路1号"，那没问题；如果是"address": "北京市朝阳区建国路1号"（中间有+），就是编码问题。

解决方案：在 Connector 的call方法里，手动控制编码：

# 不要用 httpx 自动编码 # response = await client.get(url, params=params) # 而是手动拼接 from urllib.parse import urlencode query_string = urlencode(params, safe='/', encoding='utf-8') full_url = f"{url}?{query_string}" response = await client.get(full_url)

safe='/'表示/不编码，encoding='utf-8'确保中文正确。这是处理地址、文件路径等含特殊字符参数的通用技巧。