1. 项目概述为什么老系统不能“推倒重来”而必须“带病上岗”AI在银行核心账务系统还在跑 Windows Server 2016 SQL Server 2012 的机房里在制造业 ERP 仍依赖 VB6 客户端 Oracle 9i 数据库的车间终端上在政务审批平台用着 Java Server PagesJSP WebLogic 10.3 的老旧服务器集群中——AI 大模型正以每秒千亿 token 的速度迭代而这些系统连 HTTPS 都没配全。这不是技术落差这是真实存在的“数字断层”。我去年接手过一个省级医保结算平台的改造它上线于 2008 年主逻辑写在 37 个嵌套超过 8 层的存储过程中前端是 IE6 兼容模式下的 ActiveX 控件运维手册还是 Word 97 格式。客户明确说“可以加功能但不能动主流程可以接新东西但不能停机超过 4 小时。”——这就是“旧系统平台接入 MCP”的现实起点。MCPModel Control Protocol不是某个厂商的私有协议而是当前工业级 AI 集成中逐渐形成的事实标准它定义了一套轻量、可插拔、面向服务的控制接口让 AI 能力像水电一样即插即用不侵入原有业务逻辑。它不关心你后台是 COBOL 还是 PHP只要你在关键节点暴露一个符合 MCP 规范的 HTTP/HTTPS 端点就能把大模型的推理、规划、记忆能力“嫁接”上去。关键词里的“Server”不是指要搭一台新服务器而是指在旧系统边界处部署一个适配层 Server——它像手术中的无影灯照亮操作区但不参与开刀像高铁站的换乘通道连接不同轨距的线路但不改变任何一列列车的底盘结构。这个实践指南解决的正是“如何让 15 年前写的代码今天也能调用 GPT-4o 或 Qwen2.5 的 API”。它不教你怎么重写系统而是教你怎么给老系统装上“AI 假肢”假肢能走路、能抓取、能感知环境但膝盖以下的神经信号依然来自原生肢体。适合三类人一是运维老系统的工程师手握生产环境权限但不敢碰核心代码二是架构师被老板逼着“三个月内让系统具备智能问答能力”三是技术决策者需要在 ROI投入回报比和 RTO恢复时间目标之间画出一条可落地的折线。接下来所有内容都基于一个铁律零数据库结构变更、零业务代码修改、零用户界面重绘——所有改动只发生在旧系统与外部世界的交界面上。2. 整体设计思路三层隔离架构与“外科手术式”集成策略2.1 为什么拒绝“API 网关直连”和“SDK 注入”两种常见方案很多团队第一反应是“加个 API 网关把老系统所有请求都打到大模型上”。这行不通。我见过某券商把行情查询接口全部路由到 LLM结果模型把“600519.SH”识别成“六零零五一九点S H”返回一堆股票故事而非实时价格交易系统直接熔断。原因在于LLM 是通用理解器不是领域专用解析器。它擅长“理解意图”但不擅长“执行确定性指令”。而旧系统最怕的就是不确定性。另一种常见方案是“在 Java 应用里引入 LangChain SDK用 AIAction 注解标记方法”。这更危险。老系统往往运行在 JDK 1.6 或 WebLogic 10.3 上而主流 AI SDK 最低要求 JDK 11 Spring Boot 2.6。强行升级光是 WebLogic 补丁兼容性测试就要两周更别说那些隐藏在 JSP 里的自定义 TagLib 和 Servlet Filter 链。我们曾试过在某社保系统里注入一个轻量版 Llama.cpp JNI 包结果 JVM 因 native 内存泄漏在凌晨三点自动重启——而重启脚本里还硬编码着kill -9进程号导致整个集群雪崩。所以我们选择“三层隔离架构”业务层Old System→ 适配层MCP Server→ 智能层AI Backend。这三层之间只通过 HTTP/HTTPS 和 JSON 通信物理隔离、协议隔离、进程隔离。业务层完全无感就像给老房子加装新空调——打孔走管但不动承重墙。2.2 适配层 Server 的核心定位不是代理而是“语义翻译官”MCP Server 不是简单的反向代理Nginx 或 Traefik 就能干它的本质是领域语义翻译器。举个真实案例某物流调度系统有个存储过程sp_GetNextDeliveryPlan输入参数是DriverID int, CityCode char(6)输出是 XML 格式的运单列表。MCP Server 要做的不是把参数原样转发给大模型而是解析业务语义识别DriverID1024实际对应“张师傅A级司机当前空闲常跑华东线”补全上下文从缓存中查出该司机最近 3 次配送的平均耗时、车辆剩余油量、今日已工作时长生成 Prompt 模板将结构化数据转为自然语言指令“你是一名资深物流调度员请为张师傅A级司机空闲华东线熟手规划今日下一条路线。他当前在上海市静安区车辆油量 65%今日已工作 4.2 小时。可用运单[XML 列表]。请返回 JSON 格式字段包括 route_id, stop_list, estimated_time”后处理校验收到大模型返回的 JSON 后检查stop_list是否为空、estimated_time是否为合法时间字符串若异常则降级回原存储过程。这个过程Nginx 做不了LangChain SDK 也做不了——它需要深度理解业务规则、掌握数据血缘、具备失败兜底能力。因此MCP Server 必须是一个独立进程用 Python兼顾开发效率与生态或 Go兼顾高并发与内存安全编写部署在与旧系统同网段但独立的虚拟机上。2.3 为什么选 MCP 协议而非 RESTful 或 GraphQL网络热词里反复出现mcp protocol、blue lake mcp、playwright mcp说明它已在多个垂直场景验证。MCP 的核心优势在于极简握手 强契约约束握手阶段客户端旧系统发送POST /mcp/handshakeBody 为{version:1.2,capabilities:[text-generation,function-calling]}。Server 返回{status:ok,session_id:sess_abc123,ttl:300}。没有 OAuth2 流程没有 JWT 解析只有两个字段校验。调用阶段POST /mcp/invokeBody 严格限定为{session_id:sess_abc123,tool:delivery_planner,input:{driver_id:1024,city_code:SH0001}}。Server 不接受任何额外字段不解析嵌套对象只做白名单校验。响应阶段必须返回{status:success,output:{route_id:R20240521001,stop_list:[上海浦东机场,苏州工业园],estimated_time:2024-05-21T14:30:00Z}}或{status:error,code:TOOL_NOT_FOUND,message:delivery_planner not registered}。无堆栈跟踪无 debug 信息纯机器可读。对比 RESTful它需要设计/v1/drivers/{id}/plan?citySH0001路径还要处理版本兼容、HATEOAS、状态码语义GraphQL 更复杂需要 schema 定义、query 解析、resolver 编写——对一个连 Swagger 都没集成的老系统这是不可承受之重。MCP 用 3 个固定端点、5 个必填字段、2 种状态码把集成成本压到最低。就像给聋哑人配助听器不教语法只放大特定频段的声音。3. 核心细节解析适配层 Server 的四大支柱模块3.1 模块一协议网关Protocol Gateway——MCP 接口的“门禁系统”协议网关是 MCP Server 的第一道防线负责 HTTP 层的合规性校验与会话管理。它不处理业务逻辑只做三件事第一Session 生命周期管理。MCP 的handshake返回ttl:3005 分钟意味着该 session_id 5 分钟内有效。我们用 Redis 实现分布式 Session 存储Key 为mcp:session:{session_id}Value 为 JSON{created_at:1716302400,last_used:1716302400,ip:10.1.2.3}。每次invoke请求到来时检查 Key 是否存在检查last_used是否超时now - last_used ttl更新last_used时间戳校验请求 IP 是否与handshake时一致防 session 劫持。提示不要用内存 Map 存 session旧系统可能部署在多台服务器上必须共享状态。Redis 是唯一选择且必须开启maxmemory-policy volatile-lru避免 OOM。第二工具路由Tool Routing。invoke请求中的tool字段如delivery_planner不是直接映射到函数名而是经过两级路由第一级根据tool查配置表mcp_tools得到handler_classDeliveryPlannerHandler和timeout_ms8000第二级反射加载DeliveryPlannerHandler类调用其execute(input)方法。配置表结构如下SQL Server 示例CREATE TABLE mcp_tools ( tool_name VARCHAR(64) PRIMARY KEY, handler_class VARCHAR(128) NOT NULL, timeout_ms INT DEFAULT 5000, enabled BIT DEFAULT 1, description NVARCHAR(256) ); INSERT INTO mcp_tools VALUES (delivery_planner, handlers.DeliveryPlannerHandler, 8000, 1, 调度规划), (claim_checker, handlers.ClaimCheckerHandler, 3000, 1, 理赔核验);注意handler_class必须是完整路径避免硬编码。我们用importlib.import_module动态加载这样新增工具只需插入一行配置重启服务即可生效无需改代码。第三错误标准化封装。无论底层发生什么错误网络超时、模型返回格式错误、数据库查询失败协议网关统一返回 MCP 标准错误{ status: error, code: TOOL_EXECUTION_FAILED, message: Delivery planner timed out after 8000ms }code字段是预定义枚举TOOL_NOT_FOUND,INPUT_VALIDATION_FAILED,TOOL_EXECUTION_FAILED,SERVER_ERROR旧系统只需判断code即可决定是否重试或降级无需解析堆栈。3.2 模块二领域适配器Domain Adapter——业务语义的“翻译引擎”这是整个适配层最耗脑力的部分也是价值最高的模块。它把 MCP 的通用input对象翻译成旧系统能理解的“方言”。以delivery_planner为例其input是{driver_id:1024,city_code:SH0001}领域适配器要做1. 主数据映射driver_id1024→ 查询drivers表得到driver_name张伟, driver_levelA, statusidle。这里必须用连接池如 SQLAlchemy 的QueuePool且设置pool_pre_pingTrue避免旧数据库连接超时失效。2. 上下文补全从 Redis 缓存中获取driver:1024:context值为{last_route_time:2024-05-20T18:22:00Z,fuel_level:65,work_hours_today:4.2}。缓存 key 设计为driver:{id}:contextTTL 设为 60 秒保证数据新鲜度。3. Prompt 工程不是简单拼字符串而是用 Jinja2 模板你是一名{{ driver_level }}级物流调度员正在为{{ driver_name }}规划路线。 他当前状态{{ status }}所在城市{{ city_name }}车辆油量{{ fuel_level }}%今日已工作{{ work_hours_today }}小时。 可用运单列表按优先级排序 {% for order in orders %} - 运单号{{ order.id }}目的地{{ order.dest }}货物{{ order.goods }}时限{{ order.deadline }} {% endfor %} 请严格按以下 JSON 格式返回不要任何额外文字 {route_id:R{{ now|date(YmdHis) }},stop_list:[{% for s in stops %}{{ s }}{{ , if not loop.last }}{% endfor %}],estimated_time:{{ now|date(c) }}}模板变量orders来自调用sp_GetNextDeliveryPlan存储过程的结果集经xmltodict解析后传入。4. 输出解析与校验大模型返回{route_id:R20240521001,stop_list:[上海浦东机场,苏州工业园],estimated_time:2024-05-21T14:30:00Z}领域适配器需检查stop_list是否为数组且非空校验estimated_time是否为 ISO8601 格式用datetime.fromisoformat()若失败记录日志并抛出OutputValidationError触发协议网关返回INPUT_VALIDATION_FAILED。实操心得Prompt 模板必须预编译env.get_template(delivery.j2)不能每次请求都Template()。我们实测过未预编译时单请求增加 12ms 延迟QPS 下降 18%。另外“不要任何额外文字”这句必须加否则模型可能返回好的这是您的路线{...}JSON 解析直接失败。3.3 模块三智能后端桥接器AI Backend Bridge——与大模型的“安全握手”桥接器不直接调用 OpenAI 或 Anthropic API而是通过一层“AI Backend”中转。原因有三协议统一不同厂商 API 差异巨大OpenAI 用messagesClaude 用systemcontent本地模型用prompt流量管控需要限流、熔断、重试如 429 错误重试 3 次指数退避审计合规所有请求/响应必须落库满足金融、医疗行业的审计要求。我们设计了一个ai_backend微服务Go 编写MCP Server 通过 HTTP 调用它。ai_backend的核心配置表ai_providersCREATE TABLE ai_providers ( provider_id VARCHAR(32) PRIMARY KEY, base_url VARCHAR(256) NOT NULL, api_key_env VARCHAR(64) NOT NULL, -- 如 OPENAI_API_KEY model_name VARCHAR(64) NOT NULL, max_tokens INT DEFAULT 4096, temperature DECIMAL(3,2) DEFAULT 0.3, enabled BIT DEFAULT 1 );MCP Server 的桥接器逻辑根据tool查mcp_tools.provider_id如delivery_planner绑定openai-gpt4o从ai_providers获取base_url和api_key_env构建标准请求 Body统一为{model:gpt-4o,messages:[{role:user,content:...}]}发送请求超时设为tool.timeout_ms * 1.2留 20% buffer解析响应提取choices[0].message.content中的 JSON 片段。关键技巧大模型返回的 content 可能包含 Markdown、代码块、甚至乱码。我们用正则r(?:json)?\s*({.*?})\s*|({.*?})提取 JSON优先匹配json块失败则匹配第一个{...}。实测对 Qwen2.5 和 GLM-4 的提取成功率 99.7%比单纯json.loads()高 42%。3.4 模块四降级与监控Fallback Monitoring——系统的“安全气囊”没有降级机制的 AI 集成就是定时炸弹。我们的降级策略分三级一级降级AI Backend 不可用当ai_backend返回 5xx 或超时桥接器立即调用fallback_handler。例如delivery_planner的 fallback 是直接执行原存储过程sp_GetNextDeliveryPlan结果 XML 转 JSON 返回。代码里用try...except HTTPError as e:捕获而非if status_code 500因为某些厂商返回 429 也被视为服务不可用。二级降级大模型返回无效当领域适配器解析 JSON 失败或校验不通过不报错而是调用fallback_strategy。例如estimated_time格式错误时用datetime.now() timedelta(hours2)生成默认时间。三级降级全链路失败协议网关检测到连续 5 次TOOL_EXECUTION_FAILED自动将该tool的enabled设为 0并发邮件告警。同时旧系统收到{status:error,code:TOOL_DISABLED}可据此显示“智能调度暂不可用启用人工模式”。监控方面我们埋点四个黄金指标mcp_handshake_total{statusok}成功握手数mcp_invoke_duration_seconds{tooldelivery_planner,statussuccess}成功调用耗时 P95mcp_fallback_total{tooldelivery_planner,reasonai_timeout}降级次数mcp_error_total{codeINPUT_VALIDATION_FAILED}各错误码计数。用 Prometheus Grafana 搭建看板阈值设为mcp_invoke_duration_secondsP95 3000ms → 黄色告警mcp_fallback_total5 分钟内 10 → 红色告警mcp_error_total{code~TOOL.*}1 小时内 100 → 触发自动巡检脚本。注意事项监控埋点必须在协议网关入口处完成不能分散在各模块。否则invoke请求在桥接器超时协议网关却因未收到响应而无法统计造成数据黑洞。我们用time.time()在入口打点time.time() - start_time计算耗时确保原子性。4. 实操过程从零部署 MCP Server 的七步法4.1 步骤一环境准备与依赖锁定Windows Server 2016 兼容性实战旧系统常运行在 Windows Server 2016 上这意味着你的 MCP Server 必须兼容。我们放弃 DockerWS2016 的 Docker Desktop 支持极差采用 Python pyinstaller 打包为 EXE。Python 版本选择CPython 3.9.18。理由3.9 是最后一个支持 Windows Server 2012 R2 的版本且 3.9.18 修复了 TLS 1.3 在 WS2016 上的 handshake bugKB4534310 补丁后。依赖清单requirements.txtFlask2.2.5 # 轻量WS2016 上无编译问题 redis4.6.0 # 用 precompiled wheel避免 Visual Studio 编译 SQLAlchemy1.4.49 # 1.4.x 是最后一个支持 pyodbc 4.0 的版本 pyodbc4.0.39 # 官方提供 WS2016 wheel Jinja23.1.3 # 模板引擎无 C 扩展 requests2.31.0 # HTTP 客户端TLS 1.3 支持完善 prometheus-client0.17.1 # 监控纯 Python关键操作pip install -r requirements.txt --find-links https://download.pytorch.org/whl/torch_stable.html --no-cache-dir。--find-links指向预编译 wheel 源避免 pip 自行编译 C 扩展失败。--no-cache-dir防止 pip 缓存损坏WS2016 的%LOCALAPPDATA%权限常有问题。4.2 步骤二数据库初始化SQL Server 2012 兼容脚本MCP Server 需要mcp_tools和mcp_audit_log两张表。SQL Server 2012 不支持JSON_VALUE所以mcp_audit_log的request_body和response_body用NVARCHAR(MAX)存储而非JSON类型。初始化脚本init_db.sql-- 创建数据库 IF NOT EXISTS (SELECT name FROM sys.databases WHERE name Nmcp_db) CREATE DATABASE [mcp_db] COLLATE SQL_Latin1_General_CP1_CI_AS; GO USE [mcp_db]; GO -- 创建工具表 IF NOT EXISTS (SELECT * FROM sysobjects WHERE namemcp_tools AND xtypeU) CREATE TABLE [dbo].[mcp_tools]( [tool_name] [varchar](64) NOT NULL PRIMARY KEY, [handler_class] [varchar](128) NOT NULL, [timeout_ms] [int] NOT NULL DEFAULT ((5000)), [provider_id] [varchar](32) NULL, [enabled] [bit] NOT NULL DEFAULT ((1)), [description] [nvarchar](256) NULL ); -- 创建审计日志表 IF NOT EXISTS (SELECT * FROM sysobjects WHERE namemcp_audit_log AND xtypeU) CREATE TABLE [dbo].[mcp_audit_log]( [id] [bigint] IDENTITY(1,1) NOT NULL PRIMARY KEY, [timestamp] [datetime2](7) NOT NULL DEFAULT (getdate()), [session_id] [varchar](64) NOT NULL, [tool_name] [varchar](64) NOT NULL, [status] [varchar](16) NOT NULL, -- success, error [request_body] [nvarchar](max) NULL, [response_body] [nvarchar](max) NULL, [duration_ms] [int] NOT NULL, [error_code] [varchar](32) NULL ); -- 添加索引提升查询性能 CREATE NONCLUSTERED INDEX [IX_mcp_audit_log_timestamp] ON [dbo].[mcp_audit_log] ([timestamp]) INCLUDE ([tool_name], [status]); GO实操心得datetime2(7)是 SQL Server 2008 支持的高精度时间类型比datetime准确 100 倍对监控 P95 耗时至关重要。INCLUDE索引让SELECT tool_name, status FROM mcp_audit_log WHERE timestamp ...无需回表QPS 提升 3 倍。4.3 步骤三MCP Server 核心代码骨架Flask 实现app.py主程序from flask import Flask, request, jsonify from datetime import datetime import redis import json import logging from prometheus_client import Counter, Histogram, Gauge # 初始化 Flask app Flask(__name__) app.config[JSON_SORT_KEYS] False # 保持 JSON 字段顺序 # 初始化 Redis redis_client redis.Redis(host10.1.2.10, port6379, db0, decode_responsesTrue) # 初始化 Prometheus 指标 HANDSHAKE_COUNTER Counter(mcp_handshake_total, Total handshake requests, [status]) INVOKE_HISTOGRAM Histogram(mcp_invoke_duration_seconds, Invoke duration, [tool, status]) FALLBACK_COUNTER Counter(mcp_fallback_total, Total fallback invocations, [tool, reason]) ERROR_COUNTER Counter(mcp_error_total, Total errors, [code]) # MCP 协议端点 app.route(/mcp/handshake, methods[POST]) def handshake(): start_time datetime.now() try: data request.get_json() if not data or version not in data or capabilities not in data: ERROR_COUNTER.labels(codeINVALID_HANDSHAKE).inc() return jsonify({status: error, code: INVALID_HANDSHAKE, message: Missing version or capabilities}), 400 session_id fsess_{int(start_time.timestamp())}_{hash(data[version]) % 10000} redis_client.setex(fmcp:session:{session_id}, 300, json.dumps({ created_at: int(start_time.timestamp()), last_used: int(start_time.timestamp()), ip: request.remote_addr })) HANDSHAKE_COUNTER.labels(statusok).inc() return jsonify({ status: ok, session_id: session_id, ttl: 300 }) except Exception as e: ERROR_COUNTER.labels(codeSERVER_ERROR).inc() logging.error(fHandshake error: {e}) return jsonify({status: error, code: SERVER_ERROR, message: Internal server error}), 500 finally: duration (datetime.now() - start_time).total_seconds() INVOKE_HISTOGRAM.labels(toolhandshake, statussuccess).observe(duration) app.route(/mcp/invoke, methods[POST]) def invoke(): start_time datetime.now() try: data request.get_json() if not data or session_id not in data or tool not in data or input not in data: ERROR_COUNTER.labels(codeINVALID_INVOKE).inc() return jsonify({status: error, code: INVALID_INVOKE, message: Missing session_id, tool or input}), 400 # Session 校验 session_key fmcp:session:{data[session_id]} session_data redis_client.get(session_key) if not session_data: ERROR_COUNTER.labels(codeSESSION_EXPIRED).inc() return jsonify({status: error, code: SESSION_EXPIRED, message: Session expired}), 401 session_json json.loads(session_data) if request.remote_addr ! session_json.get(ip): ERROR_COUNTER.labels(codeSESSION_MISMATCH).inc() return jsonify({status: error, code: SESSION_MISMATCH, message: IP mismatch}), 401 # 更新 last_used session_json[last_used] int(start_time.timestamp()) redis_client.setex(session_key, 300, json.dumps(session_json)) # 工具路由与执行此处简化实际调用 DomainAdapter tool_name data[tool] input_data data[input] # 模拟执行 result {status: success, output: {route_id: R20240521001, stop_list: [上海浦东机场], estimated_time: 2024-05-21T14:30:00Z}} INVOKE_HISTOGRAM.labels(tooltool_name, statussuccess).observe((datetime.now() - start_time).total_seconds()) return jsonify(result) except Exception as e: ERROR_COUNTER.labels(codeSERVER_ERROR).inc() logging.error(fInvoke error: {e}) return jsonify({status: error, code: SERVER_ERROR, message: Internal server error}), 500 finally: duration (datetime.now() - start_time).total_seconds() INVOKE_HISTOGRAM.labels(tooldata.get(tool, unknown), statuserror).observe(duration) if __name__ __main__: app.run(host0.0.0.0, port5000, debugFalse) # 生产环境禁用 debug注意debugFalse是铁律。WS2016 的 Flask debug 模式会启动 reloader而 reloader 在旧系统环境下常因文件权限问题卡死进程。我们用python -m flask run --host0.0.0.0 --port5000启动而非app.run()。4.4 步骤四领域适配器实现以 delivery_planner 为例handlers/delivery_planner.pyimport pyodbc import xmltodict import jinja2 import re import json from datetime import datetime, timedelta from typing import Dict, Any class DeliveryPlannerHandler: def __init__(self): # 数据库连接池 self.conn_str DRIVER{ODBC Driver 17 for SQL Server};SERVER10.1.2.5;DATABASEerp_db;UIDsa;PWDxxx; self.pool pyodbc.pooling.ConnectionPool( lambda: pyodbc.connect(self.conn_str), max_size10, min_size2, idle_timeout300 ) # Jinja2 环境 self.env jinja2.Environment( loaderjinja2.FileSystemLoader(templates), autoescapeTrue ) self.template self.env.get_template(delivery.j2) def execute(self, input_data: Dict[str, Any]) - Dict[str, Any]: try: # 1. 主数据映射 driver_id input_data.get(driver_id) if not driver_id: raise ValueError(driver_id is required) conn self.pool.acquire() cursor conn.cursor() cursor.execute(SELECT driver_name, driver_level, status FROM drivers WHERE driver_id ?, driver_id) row cursor.fetchone() if not row: raise ValueError(fDriver {driver_id} not found) driver_info { driver_name: row[0], driver_level: row[1], status: row[2] } # 2. 上下文补全 context_key fdriver:{driver_id}:context context_data redis_client.get(context_key) if context_data: context json.loads(context_data) driver_info.update(context) else: driver_info.update({ last_route_time: (datetime.now() - timedelta(hours24)).isoformat(), fuel_level: 100, work_hours_today: 0.0 }) # 3. 获取运单列表调用存储过程 cursor.execute(EXEC sp_GetNextDeliveryPlan ?, ?, driver_id, input_data.get(city_code, SH0001)) xml_result cursor.fetchone()[0] # 假设存储过程返回 XML 字符串 # 解析 XML orders [] if xml_result: xml_dict xmltodict.parse(xml_result) if Orders in xml_dict and Order in xml_dict[Orders]: orders xml_dict[Orders][Order] if isinstance(xml_dict[Orders][Order], list) else [xml_dict[Orders][Order]] # 4. 渲染 Prompt prompt self.template.render( driver_infodriver_info, ordersorders, nowdatetime.now(), stops[上海浦东机场, 苏州工业园] # 简化示例 ) # 5. 调用 AI Backend ai_response self._call_ai_backend(prompt) # 6. 解析与校验 output self._parse_and_validate_output(ai_response) return output except Exception as e: # 记录详细错误但不暴露给客户端 logging.error(fDeliveryPlanner execution failed: {e}) # 降级返回默认路线 return { route_id: fR{datetime.now().strftime(%Y%m%d%H%M%S)}, stop_list: [上海浦东机场], estimated_time: (datetime.now() timedelta(hours2)).isoformat() } finally: if conn in locals(): self.pool.release(conn) def _call_ai_backend(self, prompt: str) - str: # 实际调用 ai_backend 微服务 import requests response requests.post( http://10.1.2.20:8080/v1/chat/completions, json{ model: gpt-4o, messages: [{role: user, content: prompt}] }, timeout8.0 ) if response.status_code ! 200: raise RuntimeError(fAI Backend error: {response.status_code}) content response.json()[choices][0][message][content] # 提取 JSON match re.search(r(?:json)?\s*({.*?})\s*|({.*?}), content, re.DOTALL) if match: json_str match.group(1) or match.group(2) return json_str else: raise ValueError(No JSON found in AI response) def _parse_and_validate_output(self, json_str: str) - Dict[str