实际构建金融智能问答服务时Perplexity Computer 这类模块最容易被低估的不是模型调用而是数据源接入。当系统需要接入 20 个金融数据源面对的将不只是行情接口还有资讯、公告、财务指标、宏观数据和汇率利率等不同类型的数据源。它们各自有不同的鉴权方式、返回格式、更新频率和限流策略如果一上来就按“写死接口、解析 JSON”的方式堆代码后面每加一个数据源都会让主流程越来越难维护。下面从数据源规划讲起围绕适配器模式、上下文工程、Perplexity API 集成、验证与排错给出一个可落地的实现思路。这里说的 Perplexity Computer并不是某个官方产品的名称而是一个内部服务模块的代号。它的职责是接收用户查询从 20 个金融数据源中获取相关结构化数据再将数据分析结果交给 Perplexity API 生成自然语言答案。之所以要把数据源接入单独拿出来讨论是因为金融领域的接口种类多、更新快、容错要求高任何一个数据源不稳定都可能影响最终答案的可信度。提前规划好分类、权限和数据模型比写好调用代码更重要。1. 理解 Perplexity Computer 的定位和要解决的核心问题1.1 这不是“调用模型”问题而是“组织数据”问题很多团队做 AI 问答时习惯把重点放在 Prompt 和模型参数上结果模型选得再强遇到“贵州茅台今天为什么上涨”这类问题仍然可能给出过期信息。原因很简单模型没有实时金融数据也没有具体到某只股票的行情、新闻和公告上下文。Perplexity Computer 要做的事情是在问句进入大模型之前先完成一次数据检索和整理从 20 个数据源中识别哪些与当前问题相关。按照证券代码、时间范围、数据类型拉取数据。清洗字段去掉重复项合并同类信息。把结构化 JSON 转换成自然语言片段。在上下文中带上数据来源和时间戳减少模型编造概率。所以 Perplexity Computer 的价值集中在“组织数据”这一层。模型调用只是最后一步如果前面的数据源接入混乱后面 Prompt 写得再好也很难得到稳定答案。1.2 20 金融数据源可以从五个类别理解“20”听起来很多但仔细拆解后金融数据源通常可以归入几个稳定类别。每个类别内部有相似的业务语义但对接方式差异较大。数据类别典型内容更新频率对接难点实时行情股票最新价、涨跌幅、成交量、买卖五档秒级到分钟级字段标准不统一部分接口需要轮询K 线与历史行情日线、周线、月线、复权因子日级时间区间参数、复权方式容易理解错财经新闻与公告公司公告、行业新闻、监管信息分钟级到小时级文本长需要去重和相关性过滤财务指标营收、净利润、ROE、毛利率季度级不同报告期口径不同代码映射复杂宏观与汇率利率GDP、CPI、利率、汇率日级到月级指标代码各自独立单位不统一舆情分析也可以纳入新闻类别但要注意只使用公开新闻的情感倾向不涉及个人隐私和敏感信息。接入时建议每个类别先选 1 个稳定数据源跑通全链路再并行接入同类数据源。1.3 整体架构查询、适配、上下文、答案Perplexity Computer 的整体链路可以简化为用户问题 - 查询路由 - 数据源适配器层 - 行情数据源、新闻数据源、财务数据源、宏观数据源 - 数据合并与排序 - 上下文构建 - Perplexity API - 答案 - Redis / MySQL 缓存查询路由根据问题中的证券代码、数据类别和时间范围决定要调用哪些适配器。适配器层负责屏蔽外部接口差异统一返回标准化数据。上下文构建模块负责从结果中选取关键信息组装成模型能理解的文本。最后调用 Perplexity API 生成答案。这样的分层好处是当新增第 21 个数据源时只需要新增一个适配器实现类不需要改动查询路由和上下文构建逻辑。2. 数据源接入前的规划分类、权限和数据模型不能省2.1 数据源分类要同时考虑业务字段和技术特性接入数据源前先不要急着写 HTTP 调用。业务侧关心数据是什么工程侧关心怎么调、调用频率多少、返回结构如何这两部分都要登记清楚。业务维度按前面表格中的类别区分工程维度至少要考虑以下几点接口协议是 REST、WebSocket 还是文件下载。鉴权方式是 API Key、签名参数还是 OAuth。返回格式是 JSON、XML 还是 CSV。单次请求能返回多少条数据。每分钟或每天有多少调用配额。把这些信息记录在数据源元数据表里而不是散落在代码注释中。后续排查问题、评估成本、调整缓存策略时这张表是最快入口。还要定义一个通用查询对象避免每个适配器接收不同参数。下面是一个常见的DataQuery定义public class DataQuery { private ListString securityCodes; private DataType dataType; private Instant startTime; private Instant endTime; private Integer limit; private String extraParams; }所有适配器都接收DataQuery但内部可以按需使用字段。比如 K 线适配器用 startTime、endTime行情适配器只用 securityCodes。这样可以保证路由层逻辑统一。2.2 统一数据模型是降低接入成本的第一道防线假设三个数据源返回的字段分别是stockCode、symbol、code如果不在适配器层转换下游每个业务模块都要处理 20 种字段名代码会迅速失控。统一数据模型至少要包含来源、证券代码、数据类型、时间、数值和扩展字段。示例public class FinancialDataPoint { private String source; private String securityCode; private DataType dataType; private Instant timestamp; private BigDecimal value; private String currency; private MapString, Object meta; }这里有一个容易踩的坑不能把所有字段都塞进meta。核心业务字段比如价格、涨跌幅、净利润必须使用强类型字段这样编译期就能发现类型错误。meta只用来存放不影响核心计算的扩展信息比如“公告标题”“新闻摘要”“复权因子”。好的设计是“少量强类型字段 一个扩展 Map”。强类型字段用于路由、计算和过滤扩展 Map 用于保留原始上下文。2.3 鉴权、限流和配额要提前梳理成配置金融数据源的凭证通常不会只在一个环境使用。开发、测试、生产环境会使用不同账号配额也不同。把凭证写在配置类里比写在代码里安全但更推荐放在环境变量或配置中心。下面是一个application.yml中的数据源配置示例financial: datasources: stock-quote: enabled: true base-url: https://api.example.com/v1 app-key: ${STOCK_QUOTE_KEY} app-secret: ${STOCK_QUOTE_SECRET} qps: 5 timeout-ms: 3000 retry-count: 2 finance-news: enabled: true base-url: https://news.example.com/api app-key: ${FINANCE_NEWS_KEY} qps: 2 timeout-ms: 5000 retry-count: 1qps字段很重要。单个数据源通常有调用频率限制但多个适配器可能是并行执行的如果每个适配器都不限流峰值流量可能瞬间触发上游限流。接入时要把每个数据源的 QPS 配置化并用信号量或令牌桶控制实际调用频率。开发环境建议增加一个 Mock 数据源适配器返回固定样例数据。这样不消耗真实配额也能让联调流程稳定复现。3. 用适配器层接入 20 数据源而不是堆 if-else3.1 技术栈和工程骨架实现 Perplexity Computer 可以使用 Spring Boot 3.x Java 17配合 MyBatis-Plus 管理持久化Redis 管理缓存。选用 WebClient 作为 HTTP 客户端方便后面升级为异步调用。Maven 依赖示例dependencies dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-web/artifactId /dependency dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-webflux/artifactId /dependency dependency groupIdcom.baomidou/groupId artifactIdmybatis-plus-boot-starter/artifactId version3.5.5/version /dependency dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-data-redis/artifactId /dependency /dependencies引入 webflux 只是为了使用 WebClient不是要把整个服务改成响应式架构。Controller 层仍然使用 Spring MVC适配器层可以同步阻塞获取结果生产环境再根据流量决定是否下沉到异步任务。3.2 定义数据源适配器接口每个外部数据源对应一个适配器实现类接口定义如下public interface FinancialDataSourceAdapter { String getSourceCode(); ListFinancialDataPoint fetch(DataQuery query); }getSourceCode返回唯一标识比如a-stock-quote、hk-stock-quote、finance-news。fetch负责调用外部接口并返回统一模型。这种设计的核心收益是隔离变化。外部接口升级、字段调整、鉴权方式变化只影响对应适配器内部不会扩散到上层查询逻辑。3.3 行情适配器实现示例下面是一个 A 股行情适配器的示例实际项目请替换为已经获得授权的数据源接口。Component public class AStockQuoteAdapter implements FinancialDataSourceAdapter { private final WebClient webClient; private final DataSourceConfig config; public AStockQuoteAdapter(WebClient.Builder webClientBuilder, DataSourceConfig config) { this.config config; this.webClient webClientBuilder.baseUrl(config.getBaseUrl()).build(); } Override public String getSourceCode() { return a-stock-quote; } Override public ListFinancialDataPoint fetch(DataQuery query) { return webClient.get() .uri(uri - uri.path(/stock/quote) .queryParam(codes, String.join(,, query.getSecurityCodes())) .build()) .header(Authorization, Bearer config.getAppKey()) .retrieve() .bodyToMono(QuoteResponse.class) .map(response - response.toDataPoints(query)) .timeout(Duration.ofMillis(config.getTimeoutMs())) .block(); } }这里的QuoteResponse是适配器内部 DTO专门用于解析外部返回结构包含toDataPoints方法做字段映射。这样做的好处是业务代码不直接依赖外部协议。使用block()在低并发场景下没问题但如果 Perplexity Computer 需要同时拉取多个数据源建议改用异步方式避免线程池被外部接口耗死。3.4 适配器注册与动态路由Spring 会把所有FinancialDataSourceAdapter实现类收集到列表里可以通过Map结构做路由Service public class DataSourceRouter { private final MapString, FinancialDataSourceAdapter adapterMap; public DataSourceRouter(ListFinancialDataSourceAdapter adapters) { this.adapterMap adapters.stream() .collect(Collectors.toMap( FinancialDataSourceAdapter::getSourceCode, Function.identity() )); } public ListFinancialDataPoint fetch(String sourceCode, DataQuery query) { FinancialDataSourceAdapter adapter adapterMap.get(sourceCode); if (adapter null) { throw new UnsupportedDataSourceException(sourceCode); } return adapter.fetch(query); } }如果一次查询需要多个数据源可以用CompletableFuture并行调用ListCompletableFutureListFinancialDataPoint futures sourceCodes.stream() .map(code - CompletableFuture.supplyAsync( () - router.fetch(code, query), executor )) .collect(Collectors.toList()); ListFinancialDataPoint allPoints CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])) .thenApply(v - futures.stream() .map(CompletableFuture::join) .flatMap(List::stream) .collect(Collectors.toList())) .join();注意要单独配置ThreadPoolTaskExecutor不要使用并行流因为公共 ForkJoinPool 被阻塞会影响其他异步任务。3.5 接入 MyBatis-Plus 多数据源存储聚合结果如果 Perplexity Computer 需要把拉取到的行情快照写入业务库把调用日志写入日志库可以使用 MyBatis-Plus 的动态数据源功能。在application.yml中配置spring: datasource: dynamic: primary: finance strict: false datasource: finance: url: jdbc:mysql://localhost:3306/finance username: root password: ${MYSQL_PASSWORD} driver-class-name: com.mysql.cj.jdbc.Driver log: url: jdbc:mysql://localhost:3306/log_db username: root password: ${MYSQL_PASSWORD} driver-class-name: com.mysql.cj.jdbc.Driver在 Service 或 Mapper 上加DS注解即可切换数据源Service DS(finance) public class FinancialDataPointService { // 写入行情和财务数据 }多数据源会引入分布式事务、连接池管理等问题。生产环境要评估是否真的需要拆分多个物理库前期数据量不大时建议先使用单库按表分区。4. 接入 Perplexity API从结构化数据到高质量答案4.1 调用 Perplexity API 的基本请求Perplexity API 的请求格式与常见的 Chat Completions 接口类似通常包含model、messages、temperature、max_tokens等字段。具体模型名和参数以官方文档为准示例中先使用通用写法。{ model: sonar-medium-online, messages: [ { role: system, content: 你是一个金融数据分析助手。回答必须基于给定数据不要编造事实。 }, { role: user, content: 贵州茅台今天股价表现如何 } ], temperature: 0.2, max_tokens: 800 }在 Java 中调用时可以使用java.net.http.HttpClientprivate String callPerplexity(String systemPrompt, String userContent) { HttpRequest request HttpRequest.newBuilder() .uri(URI.create(perplexityConfig.getEndpoint())) .header(Authorization, Bearer perplexityConfig.getApiKey()) .header(Content-Type, application/json) .POST(BodyPublishers.ofString(buildRequestBody(systemPrompt, userContent))) .build(); try (HttpResponseString response httpClient.send(request, BodyHandlers.ofString())) { if (response.statusCode() ! 200) { throw new PerplexityApiException(response.statusCode(), response.body()); } return parseAnswer(response.body()); } }apiKey必须从配置中心或密钥管理服务读取不能写死在代码里。服务端调用也建议增加本地限流不要依赖上游限流后再临时处理。4.2 组织上下文把 JSON 换成“人可以读懂的句子”这一步是整个系统的质量关键。直接把原始 JSON 拼进 Prompt会让模型难以理解字段含义也就更容易输出错误答案。推荐做法是把每条数据折算成一句自然语言并带上来源和时间。例如请基于以下数据回答问题。数据截止时间2025-03-20 15:00:00北京时间。 行情数据来源a-stock-quote - 600519 贵州茅台最新价 1680.00 元涨跌幅 1.20%成交量 3.2 万手。 新闻数据来源finance-news - 2025-03-20 14:30某机构发布白酒行业景气度报告认为高端白酒需求稳定。 财务数据来源company-finance - 2024 年归母净利润 862.28 亿元同比 15.38%。这样的上下文有几个好处模型不需要自己猜字段名。时间戳清晰模型知道数据新鲜度。来源标识明确回答时可以引用。数据量可控避免 Prompt 太长消耗 token。如果某类数据返回了多条只保留相关性最高的 Top N。相关性可以在适配器层根据meta中的 score 字段排序也可以在上下文构建模块做过滤。4.3 缓存策略不是所有问题都适合实时拉取金融数据对实时性要求高但也不能每次都实时拉取。高频轮询会消耗配额还会造成回答不稳定。建议按三个层级设计缓存缓存层级key 设计过期时间适用场景数据源结果source:dataType:code:date30 秒到 5 分钟高频的行情、新闻上下文构建结果ctx:queryHash:code:date1 分钟到 10 分钟相同问题、相同代码的重复查询最终答案answer:question:code:date1 分钟到 30 分钟用户重复提问且不要求绝对实时使用 Redis 时可以先从缓存读取未命中再拉取和构建String contextKey ctx: sha256(queryKey); String cached redisTemplate.opsForValue().get(contextKey); if (cached ! null) { return cached; } String context buildContext(dataPoints); redisTemplate.opsForValue().set(contextKey, context, Duration.ofMinutes(5));这里的关键是 key 中必须包含“日期”或“时间戳”否则缓存会把不同日期的数据混淆答案看起来就是错的。对于需要“当前价格”的查询缓存时间不要超过 1 分钟。5. 运行验证与监控能通不等于能上线5.1 本地联调时先验证四个检查点本地跑通一个查询接口并不代表系统可以上线。启动服务后至少验证以下四个方面数据源连通性每个适配器能否独立调用成功报错时错误信息是否可读。字段映射返回的FinancialDataPoint中securityCode、timestamp、value是否与预期一致。上下文格式组装出的 Prompt 是否包含数据来源和时间。Perplexity 返回最终答案是否基于提供的上下文而不是模型自由发挥。可以先用一个最小接口做联调curl -X POST http://localhost:8080/api/query \ -H Content-Type: application/json \ -d {question:600519 最近一天股价走势如何}正常返回的答案应包含股价数据、数据时间以及明显来自前面上下文的信息。如果答案里出现“根据我的知识”这类表述说明上下文没有被正确使用。5.2 数据源健康度要通过日志和指标判断多个数据源在线的系统最怕的是某个数据源静默失败。比如行情接口返回空数组业务逻辑可能当作“没有数据”继续执行用户得到不完整的答案。每个适配器都要记录以下指标adapter_request_total带数据源名称和状态标签。adapter_request_latency_ms统计平均耗时和 P99。adapter_failure_total区分超时、限流、解析失败。context_build_duration_ms观察上下文组装耗时。perplexity_call_total记录 Perplexity API 的调用量。如果是 Prometheus Grafana可以按数据源维度看失败率。日志中至少包含requestId、source、securityCode、latency、error方便排障。生产环境建议增加告警规则某个数据源连续失败 10 次、平均耗时超过阈值、Perplexity API 429 次数突然上升都要能触达值班人员。5.3 开发环境与生产环境的差距开发环境可以用 Mock 数据源但生产环境必须补齐能力。下面列出开发与生产的主要差异环境数据源凭证缓存监控降级策略开发Mock 或真实测试账号本地环境变量不缓存或短缓存控制台日志无测试真实测试账号测试密钥Redis 独立实例完整日志手动开关生产真实生产账号密钥管理服务多级缓存指标与告警自动熔断和降级生产环境还要考虑配置外置化。数据源开关、超时时间、缓存过期时间都应该通过配置中心下发不要改代码后重新发布。6. 常见问题排查6.1 接口返回 200 但数据为空现象适配器没有抛异常但返回的列表为空最终答案缺少关键数据。可能原因请求参数没有传全比如缺少时间范围。鉴权头字段名与上游要求不一致。上游把数据放在 JSON 的data.list里解析对象只看了顶层。适配器把空白字符串当成合法值过滤掉了正常数据。检查方式在适配器中打印请求 URL 和原始响应体。用同一参数在浏览器或 API 调试工具中请求一次。对照上游文档检查 JSON 层级和字段名。处理建议在适配器内部增加“空数据告警”当返回 200 但业务数据为空时输出 warn 日志并携带原始响应片段。不要直接吞掉空结果。6.2 Perplexity API 频繁返回 429现象答案接口偶发失败日志中出现 429 或 Too Many Requests。可能原因上游 QPS 限制过低。Perplexity Computer 没有本地限流。相同问题没有走缓存每次都触发 API 调用。并发拉取多个数据源后在短时间内集中调用 Perplexity。检查方式查看每日调用量和配额对比。查看perplexity_call_total的峰值速率。确认 Redis 缓存是否生效ctx:*和answer:*的命中率。处理建议在 Perplexity 调用层加信号量限制并发例如最多 5 个并发private final Semaphore perplexitySemaphore new Semaphore(5); public String callWithLimit(String prompt) { boolean acquired perplexitySemaphore.tryAcquire(100, TimeUnit.MILLISECONDS); if (!acquired) { throw new TooManyRequestsException(Perplexity 并发超限); } try { return callPerplexity(prompt); } finally { perplexitySemaphore.release(); } }同时为短时间重复问题增加最终答案缓存降低 API 调用压力。6.3 答案里出现明显错误数据现象答案中的价格、日期、涨跌幅与原始数据源不一致或者模型自己补出不存在的数据。可能原因上下文没有带上数据时间模型无法判断数据新旧。上下文过长关键数据被截断。Prompt 没有要求“只能基于上下文回答”。多个数据源对同一字段口径不同比如前复权价格与不复权价格混用。检查方式打印最终发送给 Perplexity 的 Prompt检查是否包含时间戳、来源和完整数字。处理建议在系统提示词中明确要求“只使用给定数据不要补全”。如果某些问题无法从上下文得到答案应让模型直接说明“当前数据不足”。另外对答案中的数字做后校验如果模型输出的价格字段与上下文不一致可以标记为低置信度结果。6.4 新增第 21 个数据源时改了多处代码说明结构已经腐化现象每加一个数据源需要修改查询路由、上下文构建、存储结构甚至改动FinancialDataPoint的核心字段。根本原因适配器接口没有统一或者新数据源返回的数据无法映射到统一模型。处理建议重新梳理适配器接口确保所有实现类都通过 Spring Bean 自动注册。查询路由不要写 if-else 分支而是通过配置决定哪些数据源参与哪个问题类型。新增数据源时只需要新增实现类、配置文件、测试用例和监控项。7. 最佳实践与落地清单7.1 接入第 21 个数据源的检查清单无论是第 1 个还是第 21 个数据源接入时都按下面清单执行可以减少返工确认数据来源具有合法授权符合业务合规要求。记录数据源名称、更新频率、QPS 限额、鉴权方式。定义DataQuery参数映射保证外层不用感知差异。编写适配器实现类核心字段映射到FinancialDataPoint。在本地用测试账号跑通一个最小查询用例。配置超时、重试、限流和空结果告警。设计 Redis key并明确缓存过期时间。添加指标和日志字段确保数据源失败可观测。补充单元测试和对接文档记录字段映射示例。这个清单适用于外部三方接口也适用于内部其他团队维护的 HTTP 服务。7.2 数据源接入时最容易踩的三个坑第一个坑是凭证硬编码。把appKey和secret写在 Java 类里项目一旦被 fork 或者代码仓库泄露金融数据源账号就可能被滥用。正确做法是通过环境变量、配置中心或密钥管理服务注入代码库只保留占位符。第二个坑是不做统一模型。每个数据源解析出不同结构直接往业务层传导致查询代码充满字段映射判断。接入前要先定义FinancialDataPoint和DataQuery所有适配器共用同一套数据协议。第三个坑是上下文过载或缺少时间戳。把 1000 条 JSON 全部塞进 Prompt模型无法抓住重点不写时间戳模型会把几天前的旧数据当作当前数据回答。上下文构建必须做排序、截断和时间标注关键数据只能保留 Top N。7.3 从“接口接入”走向“数据治理”把 20 数据源稳定接入只是第一步。后续可以围绕数据质量做更多扩展为数据源增加质量评分记录单位时间内空值率和异常值率。建立数据血缘关系让每个答案都能追溯到具体数据来源。使用事件总线推送实时行情变化替代高频轮询。支持多轮对话记忆让用户追问“那最近一周呢”时不需要重新解释上下文。按用户权限控制数据源可见性避免内部数据越权访问。这些能力都会让 Perplexity Computer 从“能回答问题”进化为“可信赖的金融数据助手”。回到最初的技术判断Perplexity Computer 最值得投入的部分不是模型参数调优而是数据源的规范化接入和上下文组织。适配器、统一数据模型、缓存分层、监控告警这些看起来不性感的工作恰恰决定了最终答案是否准确、系统是否能持续接进新数据源。后续再遇到“接入第 30 个数据源”的需求时只要前面这些基础打牢剩下的只是配置和测试工作。
金融智能问答服务：数据源接入与Perplexity集成实践
实际构建金融智能问答服务时Perplexity Computer 这类模块最容易被低估的不是模型调用而是数据源接入。当系统需要接入 20 个金融数据源面对的将不只是行情接口还有资讯、公告、财务指标、宏观数据和汇率利率等不同类型的数据源。它们各自有不同的鉴权方式、返回格式、更新频率和限流策略如果一上来就按“写死接口、解析 JSON”的方式堆代码后面每加一个数据源都会让主流程越来越难维护。下面从数据源规划讲起围绕适配器模式、上下文工程、Perplexity API 集成、验证与排错给出一个可落地的实现思路。这里说的 Perplexity Computer并不是某个官方产品的名称而是一个内部服务模块的代号。它的职责是接收用户查询从 20 个金融数据源中获取相关结构化数据再将数据分析结果交给 Perplexity API 生成自然语言答案。之所以要把数据源接入单独拿出来讨论是因为金融领域的接口种类多、更新快、容错要求高任何一个数据源不稳定都可能影响最终答案的可信度。提前规划好分类、权限和数据模型比写好调用代码更重要。1. 理解 Perplexity Computer 的定位和要解决的核心问题1.1 这不是“调用模型”问题而是“组织数据”问题很多团队做 AI 问答时习惯把重点放在 Prompt 和模型参数上结果模型选得再强遇到“贵州茅台今天为什么上涨”这类问题仍然可能给出过期信息。原因很简单模型没有实时金融数据也没有具体到某只股票的行情、新闻和公告上下文。Perplexity Computer 要做的事情是在问句进入大模型之前先完成一次数据检索和整理从 20 个数据源中识别哪些与当前问题相关。按照证券代码、时间范围、数据类型拉取数据。清洗字段去掉重复项合并同类信息。把结构化 JSON 转换成自然语言片段。在上下文中带上数据来源和时间戳减少模型编造概率。所以 Perplexity Computer 的价值集中在“组织数据”这一层。模型调用只是最后一步如果前面的数据源接入混乱后面 Prompt 写得再好也很难得到稳定答案。1.2 20 金融数据源可以从五个类别理解“20”听起来很多但仔细拆解后金融数据源通常可以归入几个稳定类别。每个类别内部有相似的业务语义但对接方式差异较大。数据类别典型内容更新频率对接难点实时行情股票最新价、涨跌幅、成交量、买卖五档秒级到分钟级字段标准不统一部分接口需要轮询K 线与历史行情日线、周线、月线、复权因子日级时间区间参数、复权方式容易理解错财经新闻与公告公司公告、行业新闻、监管信息分钟级到小时级文本长需要去重和相关性过滤财务指标营收、净利润、ROE、毛利率季度级不同报告期口径不同代码映射复杂宏观与汇率利率GDP、CPI、利率、汇率日级到月级指标代码各自独立单位不统一舆情分析也可以纳入新闻类别但要注意只使用公开新闻的情感倾向不涉及个人隐私和敏感信息。接入时建议每个类别先选 1 个稳定数据源跑通全链路再并行接入同类数据源。1.3 整体架构查询、适配、上下文、答案Perplexity Computer 的整体链路可以简化为用户问题 - 查询路由 - 数据源适配器层 - 行情数据源、新闻数据源、财务数据源、宏观数据源 - 数据合并与排序 - 上下文构建 - Perplexity API - 答案 - Redis / MySQL 缓存查询路由根据问题中的证券代码、数据类别和时间范围决定要调用哪些适配器。适配器层负责屏蔽外部接口差异统一返回标准化数据。上下文构建模块负责从结果中选取关键信息组装成模型能理解的文本。最后调用 Perplexity API 生成答案。这样的分层好处是当新增第 21 个数据源时只需要新增一个适配器实现类不需要改动查询路由和上下文构建逻辑。2. 数据源接入前的规划分类、权限和数据模型不能省2.1 数据源分类要同时考虑业务字段和技术特性接入数据源前先不要急着写 HTTP 调用。业务侧关心数据是什么工程侧关心怎么调、调用频率多少、返回结构如何这两部分都要登记清楚。业务维度按前面表格中的类别区分工程维度至少要考虑以下几点接口协议是 REST、WebSocket 还是文件下载。鉴权方式是 API Key、签名参数还是 OAuth。返回格式是 JSON、XML 还是 CSV。单次请求能返回多少条数据。每分钟或每天有多少调用配额。把这些信息记录在数据源元数据表里而不是散落在代码注释中。后续排查问题、评估成本、调整缓存策略时这张表是最快入口。还要定义一个通用查询对象避免每个适配器接收不同参数。下面是一个常见的DataQuery定义public class DataQuery { private ListString securityCodes; private DataType dataType; private Instant startTime; private Instant endTime; private Integer limit; private String extraParams; }所有适配器都接收DataQuery但内部可以按需使用字段。比如 K 线适配器用 startTime、endTime行情适配器只用 securityCodes。这样可以保证路由层逻辑统一。2.2 统一数据模型是降低接入成本的第一道防线假设三个数据源返回的字段分别是stockCode、symbol、code如果不在适配器层转换下游每个业务模块都要处理 20 种字段名代码会迅速失控。统一数据模型至少要包含来源、证券代码、数据类型、时间、数值和扩展字段。示例public class FinancialDataPoint { private String source; private String securityCode; private DataType dataType; private Instant timestamp; private BigDecimal value; private String currency; private MapString, Object meta; }这里有一个容易踩的坑不能把所有字段都塞进meta。核心业务字段比如价格、涨跌幅、净利润必须使用强类型字段这样编译期就能发现类型错误。meta只用来存放不影响核心计算的扩展信息比如“公告标题”“新闻摘要”“复权因子”。好的设计是“少量强类型字段 一个扩展 Map”。强类型字段用于路由、计算和过滤扩展 Map 用于保留原始上下文。2.3 鉴权、限流和配额要提前梳理成配置金融数据源的凭证通常不会只在一个环境使用。开发、测试、生产环境会使用不同账号配额也不同。把凭证写在配置类里比写在代码里安全但更推荐放在环境变量或配置中心。下面是一个application.yml中的数据源配置示例financial: datasources: stock-quote: enabled: true base-url: https://api.example.com/v1 app-key: ${STOCK_QUOTE_KEY} app-secret: ${STOCK_QUOTE_SECRET} qps: 5 timeout-ms: 3000 retry-count: 2 finance-news: enabled: true base-url: https://news.example.com/api app-key: ${FINANCE_NEWS_KEY} qps: 2 timeout-ms: 5000 retry-count: 1qps字段很重要。单个数据源通常有调用频率限制但多个适配器可能是并行执行的如果每个适配器都不限流峰值流量可能瞬间触发上游限流。接入时要把每个数据源的 QPS 配置化并用信号量或令牌桶控制实际调用频率。开发环境建议增加一个 Mock 数据源适配器返回固定样例数据。这样不消耗真实配额也能让联调流程稳定复现。3. 用适配器层接入 20 数据源而不是堆 if-else3.1 技术栈和工程骨架实现 Perplexity Computer 可以使用 Spring Boot 3.x Java 17配合 MyBatis-Plus 管理持久化Redis 管理缓存。选用 WebClient 作为 HTTP 客户端方便后面升级为异步调用。Maven 依赖示例dependencies dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-web/artifactId /dependency dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-webflux/artifactId /dependency dependency groupIdcom.baomidou/groupId artifactIdmybatis-plus-boot-starter/artifactId version3.5.5/version /dependency dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-data-redis/artifactId /dependency /dependencies引入 webflux 只是为了使用 WebClient不是要把整个服务改成响应式架构。Controller 层仍然使用 Spring MVC适配器层可以同步阻塞获取结果生产环境再根据流量决定是否下沉到异步任务。3.2 定义数据源适配器接口每个外部数据源对应一个适配器实现类接口定义如下public interface FinancialDataSourceAdapter { String getSourceCode(); ListFinancialDataPoint fetch(DataQuery query); }getSourceCode返回唯一标识比如a-stock-quote、hk-stock-quote、finance-news。fetch负责调用外部接口并返回统一模型。这种设计的核心收益是隔离变化。外部接口升级、字段调整、鉴权方式变化只影响对应适配器内部不会扩散到上层查询逻辑。3.3 行情适配器实现示例下面是一个 A 股行情适配器的示例实际项目请替换为已经获得授权的数据源接口。Component public class AStockQuoteAdapter implements FinancialDataSourceAdapter { private final WebClient webClient; private final DataSourceConfig config; public AStockQuoteAdapter(WebClient.Builder webClientBuilder, DataSourceConfig config) { this.config config; this.webClient webClientBuilder.baseUrl(config.getBaseUrl()).build(); } Override public String getSourceCode() { return a-stock-quote; } Override public ListFinancialDataPoint fetch(DataQuery query) { return webClient.get() .uri(uri - uri.path(/stock/quote) .queryParam(codes, String.join(,, query.getSecurityCodes())) .build()) .header(Authorization, Bearer config.getAppKey()) .retrieve() .bodyToMono(QuoteResponse.class) .map(response - response.toDataPoints(query)) .timeout(Duration.ofMillis(config.getTimeoutMs())) .block(); } }这里的QuoteResponse是适配器内部 DTO专门用于解析外部返回结构包含toDataPoints方法做字段映射。这样做的好处是业务代码不直接依赖外部协议。使用block()在低并发场景下没问题但如果 Perplexity Computer 需要同时拉取多个数据源建议改用异步方式避免线程池被外部接口耗死。3.4 适配器注册与动态路由Spring 会把所有FinancialDataSourceAdapter实现类收集到列表里可以通过Map结构做路由Service public class DataSourceRouter { private final MapString, FinancialDataSourceAdapter adapterMap; public DataSourceRouter(ListFinancialDataSourceAdapter adapters) { this.adapterMap adapters.stream() .collect(Collectors.toMap( FinancialDataSourceAdapter::getSourceCode, Function.identity() )); } public ListFinancialDataPoint fetch(String sourceCode, DataQuery query) { FinancialDataSourceAdapter adapter adapterMap.get(sourceCode); if (adapter null) { throw new UnsupportedDataSourceException(sourceCode); } return adapter.fetch(query); } }如果一次查询需要多个数据源可以用CompletableFuture并行调用ListCompletableFutureListFinancialDataPoint futures sourceCodes.stream() .map(code - CompletableFuture.supplyAsync( () - router.fetch(code, query), executor )) .collect(Collectors.toList()); ListFinancialDataPoint allPoints CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])) .thenApply(v - futures.stream() .map(CompletableFuture::join) .flatMap(List::stream) .collect(Collectors.toList())) .join();注意要单独配置ThreadPoolTaskExecutor不要使用并行流因为公共 ForkJoinPool 被阻塞会影响其他异步任务。3.5 接入 MyBatis-Plus 多数据源存储聚合结果如果 Perplexity Computer 需要把拉取到的行情快照写入业务库把调用日志写入日志库可以使用 MyBatis-Plus 的动态数据源功能。在application.yml中配置spring: datasource: dynamic: primary: finance strict: false datasource: finance: url: jdbc:mysql://localhost:3306/finance username: root password: ${MYSQL_PASSWORD} driver-class-name: com.mysql.cj.jdbc.Driver log: url: jdbc:mysql://localhost:3306/log_db username: root password: ${MYSQL_PASSWORD} driver-class-name: com.mysql.cj.jdbc.Driver在 Service 或 Mapper 上加DS注解即可切换数据源Service DS(finance) public class FinancialDataPointService { // 写入行情和财务数据 }多数据源会引入分布式事务、连接池管理等问题。生产环境要评估是否真的需要拆分多个物理库前期数据量不大时建议先使用单库按表分区。4. 接入 Perplexity API从结构化数据到高质量答案4.1 调用 Perplexity API 的基本请求Perplexity API 的请求格式与常见的 Chat Completions 接口类似通常包含model、messages、temperature、max_tokens等字段。具体模型名和参数以官方文档为准示例中先使用通用写法。{ model: sonar-medium-online, messages: [ { role: system, content: 你是一个金融数据分析助手。回答必须基于给定数据不要编造事实。 }, { role: user, content: 贵州茅台今天股价表现如何 } ], temperature: 0.2, max_tokens: 800 }在 Java 中调用时可以使用java.net.http.HttpClientprivate String callPerplexity(String systemPrompt, String userContent) { HttpRequest request HttpRequest.newBuilder() .uri(URI.create(perplexityConfig.getEndpoint())) .header(Authorization, Bearer perplexityConfig.getApiKey()) .header(Content-Type, application/json) .POST(BodyPublishers.ofString(buildRequestBody(systemPrompt, userContent))) .build(); try (HttpResponseString response httpClient.send(request, BodyHandlers.ofString())) { if (response.statusCode() ! 200) { throw new PerplexityApiException(response.statusCode(), response.body()); } return parseAnswer(response.body()); } }apiKey必须从配置中心或密钥管理服务读取不能写死在代码里。服务端调用也建议增加本地限流不要依赖上游限流后再临时处理。4.2 组织上下文把 JSON 换成“人可以读懂的句子”这一步是整个系统的质量关键。直接把原始 JSON 拼进 Prompt会让模型难以理解字段含义也就更容易输出错误答案。推荐做法是把每条数据折算成一句自然语言并带上来源和时间。例如请基于以下数据回答问题。数据截止时间2025-03-20 15:00:00北京时间。 行情数据来源a-stock-quote - 600519 贵州茅台最新价 1680.00 元涨跌幅 1.20%成交量 3.2 万手。 新闻数据来源finance-news - 2025-03-20 14:30某机构发布白酒行业景气度报告认为高端白酒需求稳定。 财务数据来源company-finance - 2024 年归母净利润 862.28 亿元同比 15.38%。这样的上下文有几个好处模型不需要自己猜字段名。时间戳清晰模型知道数据新鲜度。来源标识明确回答时可以引用。数据量可控避免 Prompt 太长消耗 token。如果某类数据返回了多条只保留相关性最高的 Top N。相关性可以在适配器层根据meta中的 score 字段排序也可以在上下文构建模块做过滤。4.3 缓存策略不是所有问题都适合实时拉取金融数据对实时性要求高但也不能每次都实时拉取。高频轮询会消耗配额还会造成回答不稳定。建议按三个层级设计缓存缓存层级key 设计过期时间适用场景数据源结果source:dataType:code:date30 秒到 5 分钟高频的行情、新闻上下文构建结果ctx:queryHash:code:date1 分钟到 10 分钟相同问题、相同代码的重复查询最终答案answer:question:code:date1 分钟到 30 分钟用户重复提问且不要求绝对实时使用 Redis 时可以先从缓存读取未命中再拉取和构建String contextKey ctx: sha256(queryKey); String cached redisTemplate.opsForValue().get(contextKey); if (cached ! null) { return cached; } String context buildContext(dataPoints); redisTemplate.opsForValue().set(contextKey, context, Duration.ofMinutes(5));这里的关键是 key 中必须包含“日期”或“时间戳”否则缓存会把不同日期的数据混淆答案看起来就是错的。对于需要“当前价格”的查询缓存时间不要超过 1 分钟。5. 运行验证与监控能通不等于能上线5.1 本地联调时先验证四个检查点本地跑通一个查询接口并不代表系统可以上线。启动服务后至少验证以下四个方面数据源连通性每个适配器能否独立调用成功报错时错误信息是否可读。字段映射返回的FinancialDataPoint中securityCode、timestamp、value是否与预期一致。上下文格式组装出的 Prompt 是否包含数据来源和时间。Perplexity 返回最终答案是否基于提供的上下文而不是模型自由发挥。可以先用一个最小接口做联调curl -X POST http://localhost:8080/api/query \ -H Content-Type: application/json \ -d {question:600519 最近一天股价走势如何}正常返回的答案应包含股价数据、数据时间以及明显来自前面上下文的信息。如果答案里出现“根据我的知识”这类表述说明上下文没有被正确使用。5.2 数据源健康度要通过日志和指标判断多个数据源在线的系统最怕的是某个数据源静默失败。比如行情接口返回空数组业务逻辑可能当作“没有数据”继续执行用户得到不完整的答案。每个适配器都要记录以下指标adapter_request_total带数据源名称和状态标签。adapter_request_latency_ms统计平均耗时和 P99。adapter_failure_total区分超时、限流、解析失败。context_build_duration_ms观察上下文组装耗时。perplexity_call_total记录 Perplexity API 的调用量。如果是 Prometheus Grafana可以按数据源维度看失败率。日志中至少包含requestId、source、securityCode、latency、error方便排障。生产环境建议增加告警规则某个数据源连续失败 10 次、平均耗时超过阈值、Perplexity API 429 次数突然上升都要能触达值班人员。5.3 开发环境与生产环境的差距开发环境可以用 Mock 数据源但生产环境必须补齐能力。下面列出开发与生产的主要差异环境数据源凭证缓存监控降级策略开发Mock 或真实测试账号本地环境变量不缓存或短缓存控制台日志无测试真实测试账号测试密钥Redis 独立实例完整日志手动开关生产真实生产账号密钥管理服务多级缓存指标与告警自动熔断和降级生产环境还要考虑配置外置化。数据源开关、超时时间、缓存过期时间都应该通过配置中心下发不要改代码后重新发布。6. 常见问题排查6.1 接口返回 200 但数据为空现象适配器没有抛异常但返回的列表为空最终答案缺少关键数据。可能原因请求参数没有传全比如缺少时间范围。鉴权头字段名与上游要求不一致。上游把数据放在 JSON 的data.list里解析对象只看了顶层。适配器把空白字符串当成合法值过滤掉了正常数据。检查方式在适配器中打印请求 URL 和原始响应体。用同一参数在浏览器或 API 调试工具中请求一次。对照上游文档检查 JSON 层级和字段名。处理建议在适配器内部增加“空数据告警”当返回 200 但业务数据为空时输出 warn 日志并携带原始响应片段。不要直接吞掉空结果。6.2 Perplexity API 频繁返回 429现象答案接口偶发失败日志中出现 429 或 Too Many Requests。可能原因上游 QPS 限制过低。Perplexity Computer 没有本地限流。相同问题没有走缓存每次都触发 API 调用。并发拉取多个数据源后在短时间内集中调用 Perplexity。检查方式查看每日调用量和配额对比。查看perplexity_call_total的峰值速率。确认 Redis 缓存是否生效ctx:*和answer:*的命中率。处理建议在 Perplexity 调用层加信号量限制并发例如最多 5 个并发private final Semaphore perplexitySemaphore new Semaphore(5); public String callWithLimit(String prompt) { boolean acquired perplexitySemaphore.tryAcquire(100, TimeUnit.MILLISECONDS); if (!acquired) { throw new TooManyRequestsException(Perplexity 并发超限); } try { return callPerplexity(prompt); } finally { perplexitySemaphore.release(); } }同时为短时间重复问题增加最终答案缓存降低 API 调用压力。6.3 答案里出现明显错误数据现象答案中的价格、日期、涨跌幅与原始数据源不一致或者模型自己补出不存在的数据。可能原因上下文没有带上数据时间模型无法判断数据新旧。上下文过长关键数据被截断。Prompt 没有要求“只能基于上下文回答”。多个数据源对同一字段口径不同比如前复权价格与不复权价格混用。检查方式打印最终发送给 Perplexity 的 Prompt检查是否包含时间戳、来源和完整数字。处理建议在系统提示词中明确要求“只使用给定数据不要补全”。如果某些问题无法从上下文得到答案应让模型直接说明“当前数据不足”。另外对答案中的数字做后校验如果模型输出的价格字段与上下文不一致可以标记为低置信度结果。6.4 新增第 21 个数据源时改了多处代码说明结构已经腐化现象每加一个数据源需要修改查询路由、上下文构建、存储结构甚至改动FinancialDataPoint的核心字段。根本原因适配器接口没有统一或者新数据源返回的数据无法映射到统一模型。处理建议重新梳理适配器接口确保所有实现类都通过 Spring Bean 自动注册。查询路由不要写 if-else 分支而是通过配置决定哪些数据源参与哪个问题类型。新增数据源时只需要新增实现类、配置文件、测试用例和监控项。7. 最佳实践与落地清单7.1 接入第 21 个数据源的检查清单无论是第 1 个还是第 21 个数据源接入时都按下面清单执行可以减少返工确认数据来源具有合法授权符合业务合规要求。记录数据源名称、更新频率、QPS 限额、鉴权方式。定义DataQuery参数映射保证外层不用感知差异。编写适配器实现类核心字段映射到FinancialDataPoint。在本地用测试账号跑通一个最小查询用例。配置超时、重试、限流和空结果告警。设计 Redis key并明确缓存过期时间。添加指标和日志字段确保数据源失败可观测。补充单元测试和对接文档记录字段映射示例。这个清单适用于外部三方接口也适用于内部其他团队维护的 HTTP 服务。7.2 数据源接入时最容易踩的三个坑第一个坑是凭证硬编码。把appKey和secret写在 Java 类里项目一旦被 fork 或者代码仓库泄露金融数据源账号就可能被滥用。正确做法是通过环境变量、配置中心或密钥管理服务注入代码库只保留占位符。第二个坑是不做统一模型。每个数据源解析出不同结构直接往业务层传导致查询代码充满字段映射判断。接入前要先定义FinancialDataPoint和DataQuery所有适配器共用同一套数据协议。第三个坑是上下文过载或缺少时间戳。把 1000 条 JSON 全部塞进 Prompt模型无法抓住重点不写时间戳模型会把几天前的旧数据当作当前数据回答。上下文构建必须做排序、截断和时间标注关键数据只能保留 Top N。7.3 从“接口接入”走向“数据治理”把 20 数据源稳定接入只是第一步。后续可以围绕数据质量做更多扩展为数据源增加质量评分记录单位时间内空值率和异常值率。建立数据血缘关系让每个答案都能追溯到具体数据来源。使用事件总线推送实时行情变化替代高频轮询。支持多轮对话记忆让用户追问“那最近一周呢”时不需要重新解释上下文。按用户权限控制数据源可见性避免内部数据越权访问。这些能力都会让 Perplexity Computer 从“能回答问题”进化为“可信赖的金融数据助手”。回到最初的技术判断Perplexity Computer 最值得投入的部分不是模型参数调优而是数据源的规范化接入和上下文组织。适配器、统一数据模型、缓存分层、监控告警这些看起来不性感的工作恰恰决定了最终答案是否准确、系统是否能持续接进新数据源。后续再遇到“接入第 30 个数据源”的需求时只要前面这些基础打牢剩下的只是配置和测试工作。