1. 多智能体项目里配置散落一地的真实痛点做 A2A 协议的多智能体项目最先崩的往往不是 Agent 之间的通信而是配置。我见过太多旅行规划类的多智能体 Demo天气代理、票务代理、订票代理各自维护一份api_key路由服务器再抄一份前端 Streamlit 又硬编码一份。等到要换模型、换 Key、换数据库环境就得全局搜索替换漏一处就报 401。这个基于 A2A 协议的智能助手项目目标是解决旅行规划里的信息整合难题用户输入“成都天气”或“成都到西安火车票”系统通过 LLM 路由服务器做意图识别把请求分发给天气代理服务器或票务代理服务器代理再通过 MCP 工具服务器查数据库结果回显到 Streamlit 前端。整条链路涉及 LLM 路由、天气代理、票务代理、订票代理、MCP 工具服务器、数据采集脚本和前端客户端模块一多配置模块和数据模块就成了项目骨架的地基。这一篇聚焦两件事一是把项目架构图讲清楚让你知道每个模块在链路里的位置二是把配置模块config.py、create_logger.py和数据模块SQL 建表、天气采集脚本落地并且用 TaoToken 统一 Key 和 API 通道让所有智能体服务共用一套鉴权配置。适合正在搭多智能体骨架、被多份 Key 管理折磨的开发者。TaoToken 在这里扮演的是统一模型接入层的角色把原本散落在各代理里的模型调用收敛到一个 Base URL 和一把 Key 上。2. TaoToken 统一接入前置准备多智能体鉴权收敛多智能体项目最怕的就是每个 Agent 一套模型配置。天气代理用一家、票务代理用另一家、路由服务器再换一家调试时光是核对哪把 Key 对应哪个服务就要花半天。TaoToken 的思路是把这些收敛成一套一个 Base URL、一把 API Key、一个 Model ID所有代理服务器和路由服务器都从这里读配置。先到官网注册并进入控制台在 API Keys 页面创建一把 Key。地址是 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 创建后复制保存后面config.py里要用。模型 ID 可以在模型对话页面确认比如Qwen/Qwen2.5-72B-Instruct这类通用对话模型路由服务器做意图识别用它就够了。模型对话入口https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 。Base URL 统一填https://taotoken.net/api注意这个地址不带任何查询参数是纯粹的 API 端点。如果你用的是 OpenAI 兼容的 SDK直接把base_url指向它即可如果用 LangChain 的ChatOpenAI也是同样的填法。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里面有各语言 SDK 的示例遇到参数不确定时对照一下。这里有个关键点多智能体项目里路由服务器和各个代理服务器都要调模型但它们不应该各自持有 Key。正确做法是让所有服务都从同一个config.py读取base_url、api_key、model_name这样换 Key 只改一处。这也是为什么配置模块要放在项目最底层被所有脚本依赖。如果你后面要跑长期的编码类 Agent 任务可以考虑 Coding Plan入口在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 不过本篇的旅行规划 Demo 用按量 API 就够了。3. 可复制的配置模块与数据模块落地配置模块的核心是config.py它定义了一个Config类把大模型配置、数据库配置、日志配置、意图映射全部集中管理。项目根目录下建config.py内容如下import os # 项目根目录 project_root os.path.join(os.path.dirname(os.path.abspath(__file__)), ..) # 环境切换prod / test / dev / pre_prod env test class Config: def __init__(self): # 大模型配置统一走 TaoToken self.base_url https://taotoken.net/api self.api_key sk-你的TaoToken密钥 self.model_name Qwen/Qwen2.5-72B-Instruct # 数据库配置 self.host localhost self.user root self.password 123456 self.database travel_rag # 日志配置 self.log_file os.path.join(project_root, Travel, logs/app.log) # 票务查询接口地址 self.url_123 # 意图到代理的映射 self.intent { weather: WeatherQueryAssistant, flight: TicketQueryAssistant, train: TicketQueryAssistant, concert: TicketQueryAssistant, order: TicketOrderAssistant } self.temperature 0.1 def get_mysql_config(self, env): if env prod: self.host, self.user, self.password localhost, root, root elif env dev: self.host, self.user, self.password localhost1, root1, root1 elif env test: self.host, self.user, self.password localhost2, root2, root2 else: self.host, self.user, self.password localhost3, root3, root3 return self.host, self.user, self.password, self.database if __name__ __main__: print(Config().log_file) print(Config().get_mysql_config(env))intent这个字典是路由服务器的关键它把用户意图映射到对应的 A2A 代理名称。路由服务器识别出“weather”就转发给WeatherQueryAssistant识别出“train”就转发给TicketQueryAssistant。这样新增意图类型时只改这一处映射不用动路由逻辑。日志模块create_logger.py负责全局统一的 logger避免每个模块自己printimport logging import os from config import Config def setup_logger(name, log_filelogs/app.log): os.makedirs(os.path.dirname(log_file), exist_okTrue) logger logging.getLogger(name) logger.setLevel(logging.DEBUG) # 防止重复输出的关键 logger.propagate False formatter logging.Formatter(%(name)s - %(asctime)s - %(levelname)s - %(message)s) console_handler logging.StreamHandler() console_handler.setFormatter(formatter) console_handler.setLevel(logging.INFO) file_handler logging.FileHandler(filenamelog_file, encodingutf-8, modea) file_handler.setFormatter(formatter) file_handler.setLevel(logging.DEBUG) if not logger.handlers: logger.addHandler(console_handler) logger.addHandler(file_handler) return logger logger setup_logger(Travel, Config().log_file)logger.propagate False这行必须加。因为 FastAPI 之类的框架自带日志对象如果不关掉向上传播同一条日志会在控制台打印两遍排查问题时非常干扰。数据模块分两部分SQL 建表和天气采集。建表脚本sql/sql_data.sql创建travel_rag库和四张表——train_tickets、flight_tickets、concert_tickets、weather_data。以天气表为例字段覆盖了预报日期、日出日落、温度、风向风力、湿度、气压、能见度等唯一键是(city, fx_date)保证同一城市同一天只有一条记录CREATE TABLE IF NOT EXISTS weather_data ( id INT AUTO_INCREMENT PRIMARY KEY, city VARCHAR(50) NOT NULL COMMENT 城市名称, fx_date DATE NOT NULL COMMENT 预报日期, temp_max INT COMMENT 最高温度, temp_min INT COMMENT 最低温度, text_day VARCHAR(20) COMMENT 白天天气描述, text_night VARCHAR(20) COMMENT 夜间天气描述, humidity INT COMMENT 相对湿度 (%), precip DECIMAL(5,1) COMMENT 降水量 (mm), update_time DATETIME COMMENT 数据更新时间, UNIQUE KEY unique_city_date (city, fx_date) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COLLATEutf8mb4_unicode_ci COMMENT天气数据表;票务三张表结构类似都有出发城市、到达城市、出发到达时间、班次号、座位类型、总座位、剩余座位、价格唯一键用(departure_time, train_number)这类组合防止重复插入。建库时用utf8mb4字符集和utf8mb4_unicode_ci校对规则中文城市名和艺人名才不会乱码。天气采集脚本utils/spider_weather.py从和风天气 API 拉 30 天预报写入数据库。核心函数是fetch_weather_data、store_weather_data、update_weather。请求时带Accept-Encoding: gzip响应如果是 gzip 压缩就解压再解析def fetch_weather_data(city, location): headers { X-QW-Api-Key: API_KEY, Accept-Encoding: gzip } url f{BASE_URL}?location{location} try: response requests.get(url, headersheaders, timeout10) response.raise_for_status() if response.headers.get(Content-Encoding) gzip: data gzip.decompress(response.content).decode(utf-8) else: data response.text return json.loads(data) except requests.RequestException as e: print(f请求 {city} 天气数据失败: {e}) return None写入用INSERT ... ON DUPLICATE KEY UPDATE配合唯一键实现“有则更新、无则插入”避免重复数据。定时更新用schedule库每天固定时间跑一次update_weather。这里注意时区脚本里用pytz.timezone(Asia/Shanghai)统一时区否则update_time比较会出错。4. 验证请求与成功结果多智能体链路跑通配置和数据都就位后先验证模型通道是否通。写一个最小测试脚本用 OpenAI SDK 指向 TaoTokenfrom openai import OpenAI from config import Config cfg Config() client OpenAI(base_urlcfg.base_url, api_keycfg.api_key) resp client.chat.completions.create( modelcfg.model_name, messages[{role: user, content: 成都今天天气怎么样}], temperaturecfg.temperature ) print(resp.choices[0].message.content)跑通后应该能看到模型返回的天气相关回复。这一步确认了 Base URL、Key、Model ID 三件套正确。接着验证数据库。执行建表脚本后跑一次天气采集python utils/spider_weather.py控制台会打印每个城市的写入日志类似成都 2026-07-06 数据写入/更新成功: 晴, 影响行数: 1。然后查库确认SELECT city, fx_date, text_day, temp_max, temp_min FROM weather_data WHERE city 成都 LIMIT 5;能看到成都未来几天的天气记录说明数据模块通了。最后验证 A2A 链路。启动 MCP 工具服务器mcp_weather_server.py、mcp_ticket_server.py、mcp_order_server.py再启动代理服务器weather_server.py、ticket_server.py、order_server.py最后启动客户端main.py。在 Streamlit 界面输入“成都到西安火车票”路由服务器识别出train意图转发给TicketQueryAssistant票务代理生成 SQL 查 MCP 数据库工具返回车次列表。如果界面能正常显示 G870、G655 这些车次和票价说明整条 A2A 通信链路跑通了。验证时重点看代理服务器的日志确认它拿到的base_url是https://taotoken.net/api而不是某个硬编码的旧地址。多智能体项目里只要有一个代理漏改了配置就会出现部分请求成功、部分 401 的诡异现象。5. 本篇常见错误排查401、local proxy failed 与 choices 读取失败第一个高频错误是 401 Unauthorized。报错长这样openai.AuthenticationError: Error code: 401 - {error: {message: Invalid API key}}原因通常是config.py里的api_key没换成 TaoToken 的 Key或者 Key 复制时带了空格。检查方法是打印cfg.api_key[:8]看前缀对不对确认没有多余空白。另一个可能是 Base URL 写成了带路径的形式比如https://taotoken.net/api/v1正确写法是https://taotoken.net/apiSDK 会自己拼/v1/chat/completions。第二个错误是local proxy failed或连接超时。这类报错一般是网络层问题检查本机是否能正常访问https://taotoken.net/api。如果公司网络有出口限制换一个网络环境测试。注意不要在任何配置里写代理地址TaoToken 是直连的 API 通道不需要额外代理设置。第三个错误是读取choices失败AttributeError: NoneType object has no attribute choices或者KeyError: choices。这通常发生在模型返回了错误响应但代码没做异常处理。比如请求参数里model填了一个不存在的模型 IDAPI 返回错误 JSON没有choices字段。解决方法是先打印完整响应resp client.chat.completions.create(...) print(resp.model_dump())确认model字段和 TaoToken 控制台里列出的模型 ID 完全一致。路由服务器和代理服务器如果用了不同的模型 ID也要逐个核对。第四个错误是 OAuth 相关的鉴权失败。如果你在项目里用了某些需要 OAuth 流程的组件报错会提示 token 过期或 scope 不足。TaoToken 的 API Key 是静态密钥不涉及 OAuth 刷新所以遇到 OAuth 报错基本是项目里其他第三方服务的配置问题和模型通道无关分开排查即可。第五个错误是数据库连接失败mysql.connector.errors.InterfaceError。检查config.py里get_mysql_config(env)返回的 host、user、password 是否和你本地 MySQL 一致。env变量在文件顶部测试环境默认走localhost2如果你本地 MySQL 在localhost要把env改成prod或者直接改test分支的配置。6. 多智能体项目骨架搭好之后配置模块和数据模块落地后这个 A2A 多智能体项目的骨架就立起来了。路由服务器、天气代理、票务代理、订票代理都从同一份config.py读模型配置共用 TaoToken 的 Base URL 和 Key换模型或换 Key 只改一处。数据侧SQL 建表定义了清晰的数据结构天气采集脚本定时刷新数据MCP 工具服务器查库时拿到的都是新鲜数据。下一步就是实现各个 MCP 服务器和 A2A 代理服务器的具体逻辑把天气查询、票务查询、订票三条链路分别打通。到那时你会发现前期把配置收敛好后面写代理逻辑时几乎不用再碰鉴权代码专注在业务逻辑上就行。接入文档和 API Keys 页面建议收藏后面加新代理时随时对照。