AI量化交易最近讨论最多的方向已经从“用AI写策略”变成了“让AI直接读图并给出交易分析”。所谓多个AI同时看一张图是指把同一张K线图、分时图或技术指标图同时发送给多个大语言模型让每个模型独立输出趋势判断、关键点位和风险提示再由系统汇总成一份综合结论。这种做法的价值不在于某一个模型有多强而在于多个模型在同一张图上互相印证、彼此纠偏。标题中提到的孙越AI正是这类多模型量化分析工具的一个产品化形态。下面用一套最小可运行实现拆解它的核心链路图表预处理、多模型并行调用、结果解析和综合汇总并给出本地验证和生产落地时最容易踩的坑。1. 多AI看一张图先想清楚它解决什么问题1.1 单模型分析量化图表的局限性传统量化分析有两种常见做法。一种靠人工盯盘看K线形态、均线位置、成交量变化然后凭经验判断。另一种靠规则引擎把“MACD金叉”“突破20日均线”这类条件写成程序满足条件就触发信号。前者的问题是主观、容易疲劳同一个人在不同时间看同一张图结论都可能不一样。后者的问题更隐蔽规则引擎只能识别写死的形态无法理解形态背后的市场氛围也没有办法解释“为什么这里可能是支撑位”。大模型出现后很多人尝试让模型直接看K线截图用自然语言输出分析。效果比预想好但单独用一个模型做这件事会暴露出四类问题视觉理解偏差。模型对K线实体、影线、均线交叉、成交量放量的理解不一定准确经常会把长上影线看成强势突破。上下文遗漏。K线图信息密度高模型在长上下文里可能漏掉关键位置比如前期高点、密集成交区。过度自信。即使模型判断错了它也会用很确定的语气输出“大概率上涨”缺少自我怀疑。风格偏置。同一个任务有的模型习惯看多有的模型习惯看空单模型会导致结论体系性偏移。所以“多个AI同时看一张图”不是炫技而是用模型组合来对冲单模型的不确定性。1.2 多模型并行分析的产品形态把多个AI同时看一张图拆开看本质是“多智能体评审”。系统把同一张图同时分发给多个大模型每个模型独立输出结构化报告再由汇总模块生成综合结论。产品端呈现通常是这样用户上传一张K线截图或分时图。系统显示多个模型的分析卡片每个卡片包含趋势判断、置信度、关键点位、风险点。顶部显示汇总结论标出哪些模型观点一致哪些模型存在分歧。这里的核心并不在于“谁对谁错”而在于两点共识部分是否稳定分歧部分能否提示风险。如果三个模型都看多结论可信度就比单个模型看多高得多。如果两个看多、一个看空系统应该把这个分歧明确提示出来而不是强行取平均。这里的预期收益不是“准确率提升几个点”而是风险表达更完整单模型漏掉的风险另一个模型可能会补上。1.3 这个功能适合用在什么场景并不是所有量化场景都适合让AI看图表。适合的是那些“需要快速生成分析草稿、需要多视角解读、需要人工复核”的场景。场景输入输出人工介入程度个人看盘辅助K线截图多模型分析报告中用户自己判断策略研究历史K线片段形态识别和特征汇总高研究员需复核行情日报自动生成当日截图多空观点汇总中编辑审核后发布AI教学演示任意图表不同模型的解读对比低展示差异为主自动下单实时行情图买卖信号极高不建议完全自动化最后一行尤其重要。现阶段让多个AI看一张图适合作为“分析助手”不适合直接接管交易。原因很简单模型没有实时行情权限视觉推理本身存在误差而且它无法感知盘口深度、资金流向、突发消息等非图表信息。2. 系统设计与技术拆分2.1 整体处理链路图表输入、模型分发、结果汇总多AI看一张图技术链路可以拆成五步接收用户上传的图片。对图片做统一预处理裁剪多余区域、缩放分辨率、转base64编码。把同一张图片和同一份提示词并发发送给多个模型。每个模型返回自然语言或JSON文本解析成统一结构。汇总模块对多个结构化结果做投票、合并和风险提示最后生成报告。这条链路最容易被忽略的环节是第一步和第二步。很多实现把用户上传的原图直接发给模型图片分辨率过高会导致token消耗暴涨过低又会导致模型看不清K线形态。后面会单独讲预处理参数。分发环节要注意“并发”和“超时”。多模型串行调用总耗时等于所有模型耗时相加体验会很差。正确做法是并发调用同时给每个模型设置独立的超时时间避免某个模型卡住拖垮整个任务。汇总环节不是简单拼接。如果只是把多个模型的文字结果堆在一起用户还得自己对比体验和看多个网页没有区别。汇总必须解决“观点是否一致”“分歧在哪里”“最终建议是什么”这三个问题。2.2 模型供应商层用兼容接口屏蔽差异不同模型厂商的协议不统一如果为每个模型写一套单独调用代码维护成本会很高。实践中建议统一使用OpenAI兼容协议。当前主流模型的API服务多数都提供OpenAI兼容的endpoint只需要把base_url改成对应厂商的地址同时替换api_key和model名称即可。在设计模型供应商层时要抽象出一个基础接口方法作用输入输出analyze_chart核心分析入口图片base64、文本提示词模型原始输出文本parse_result解析模型输出原始文本结构化字典health_check连通性检测无是否可用这样做的好处是后续接入新模型时不需要改动上层调度和汇总逻辑只需要新增一个配置项或一个Provider类。接入模型时需要确认这类信息先看模型是否支持图像输入vision能力再看接口地址和鉴权方式最后确认单张图片的token上限。写配置时不要照搬网上的base_url要以对应服务商当前文档为准。2.3 提示词与输出协议设计多模型分析成败的另一个关键点是输出协议。如果让模型自由发挥返回的可能是长段落也可能是列表汇总程序很难解析。建议在提示词中强制规定JSON输出结构{ trend: bullish, confidence: 0.7, key_levels: [ {price: 100.5, type: support}, {price: 105.0, type: resistance} ], risk_points: [ 成交量萎缩上涨动力不足 ], action_suggestion: 短期持有跌破支撑位减仓, analysis_detail: 均线多头排列但MACD红柱缩短 }各字段含义如下trend多空判断枚举值bullish、bearish、neutral、uncertain。confidence模型对自己判断的置信度0到1之间。key_levels关键支撑和压力位列表。risk_points风险点列表字符串数组。action_suggestion交易动作建议保留自然语言便于展示。analysis_detail分析理由用于生成详细报告。统一结构之后无论模型来自哪家厂商汇总模块都只需要处理同一套字典格式。3. 环境准备与依赖3.1 版本和依赖清单下面的实现基于Python 3.10本地开发可以直接用venv隔离环境。创建虚拟环境python3 -m venv venv source venv/bin/activate安装依赖前先确认pip版本和Python版本一致。python --version pip --version然后在项目根目录创建requirements.txtopenai1.35.0 Pillow10.0.0 requests2.31.0 PyYAML6.0.1 matplotlib3.8.0逐项说明用途依赖用途openai调用OpenAI兼容接口新版客户端统一用chat.completionsPillow图片缩放、格式转换、base64编码requests健康检查或调试接口PyYAML读取模型配置文件matplotlib生成样例K线图方便本地验证安装后确认版本特别要注意openai库版本。1.0前后API差异很大旧项目的写法是openai.ChatCompletion.create新版写法是client.chat.completions.create。3.2 项目目录结构建议按模块拆分方便后续扩展multi_ai_chart/ ├── config.yaml ├── requirements.txt ├── main.py ├── providers/ │ ├── __init__.py │ ├── base.py │ └── openai_compatible.py ├── core/ │ ├── __init__.py │ ├── image_utils.py │ └── aggregator.py └── sample/ └── kline_sample.py各文件职责文件职责main.py入口读取配置、加载图片、调度模型、汇总输出providers/base.py模型供应商抽象基类providers/openai_compatible.py兼容接口实现core/image_utils.py图片预处理与base64编码core/aggregator.py多模型结果汇总sample/kline_sample.py生成样例K线图config.yaml模型列表和参数配置这种结构在本地开发时够用上线时可以把providers和core拆成独立服务。3.3 配置文件在项目根目录创建config.yamlmodels: - name: qwen-vl-plus display_name: Qwen-VL base_url: https://dashscope.aliyuncs.com/compatible-mode/v1 api_key_env: DASHSCOPE_API_KEY model: qwen-vl-plus max_image_size: 1024 timeout: 60 - name: glm-4v-plus display_name: GLM-4V base_url: https://open.bigmodel.cn/api/paas/v4 api_key_env: ZHIPU_API_KEY model: glm-4v-plus max_image_size: 1024 timeout: 60 parallel: max_workers: 4 timeout: 90 output: language: zh-CN这里的关键参数说明参数含义推荐值影响max_image_size图片最长边像素1024过大消耗token过小模型看不清timeout单个模型请求超时60秒视觉模型推理较慢不能设太短max_workers并发线程数模型数量并发太高容易触发限流api_key_envAPI Key所在环境变量名自定义不要把明文Key写进配置注意不同厂商的base_url会变化落地前应以对应服务商当前文档为准。上述地址是常见的兼容接口格式只用于示例。4. 核心代码实现4.1 图表预处理裁剪、缩放、转Base64图片预处理是很容易被忽略的一步。用户上传的截图可能是整个屏幕包含广告栏、标题、水印这些噪声会干扰模型识别也会浪费token。建议处理顺序是打开图片。可选裁剪去掉上下边缘非图表区域。按最长边缩放到max_image_size。转成RGB模式避免RGBA图片带透明通道导致编码异常。压缩为JPEG格式质量设为90。转base64字符串。实现core/image_utils.py中的核心函数import base64 from io import BytesIO from PIL import Image def preprocess_image( image_path: str, max_size: int 1024, quality: int 90, ) - str: 读取图片缩放并编码为base64字符串。 Args: image_path: 本地图片路径。 max_size: 最长边像素超过则等比缩放。 quality: JPEG压缩质量0-100。 Returns: base64编码的JPEG图片。 with Image.open(image_path) as img: img img.convert(RGB) w, h img.size max_side max(w, h) if max_side max_size: ratio max_size / max_side new_w int(w * ratio) new_h int(h * ratio) img img.resize((new_w, new_h), Image.LANCZOS) buffer BytesIO() img.save(buffer, formatJPEG, qualityquality) encoded base64.b64encode(buffer.getvalue()).decode(utf-8) return encoded这段代码有三个关键点。第一缩放而不是裁剪。等比缩放不会改变K线形状模型对实体、影线、均线方向仍然能识别。第二转RGB再保存JPEG。PNG图片带透明通道时直接编码可能导致部分服务端解析异常转RGB后更稳。第三JPEG质量90能平衡清晰度和体积。K线图是矢量特征明显的图形质量降到85以下时细小的均线标记可能变糊。4.2 多模型客户端封装在providers/base.py中定义抽象基类from abc import ABC, abstractmethod from typing import Dict class BaseProvider(ABC): 模型供应商抽象基类。 def __init__(self, name: str): self.name name abstractmethod def analyze_chart(self, image_base64: str, prompt: str) - str: 输入图片和提示词返回模型原始文本。 raise NotImplementedError abstractmethod def parse_result(self, raw: str) - Dict: 把模型返回的原始文本解析成统一结构。 raise NotImplementedError在providers/openai_compatible.py中实现OpenAI兼容协议import json import os import re from openai import OpenAI from providers.base import BaseProvider class OpenAICompatibleProvider(BaseProvider): OpenAI兼容协议模型供应商。 def __init__( self, name: str, api_key_env: str, base_url: str, model: str, timeout: int 60, temperature: float 0.2, ): super().__init__(name) api_key os.getenv(api_key_env) if not api_key: raise ValueError(f环境变量 {api_key_env} 未设置) self.client OpenAI(api_keyapi_key, base_urlbase_url, timeouttimeout) self.model model self.temperature temperature def analyze_chart(self, image_base64: str, prompt: str) - str: response self.client.chat.completions.create( modelself.model, temperatureself.temperature, messages[ { role: user, content: [ {type: text, text: prompt}, { type: image_url, image_url: { url: fdata:image/jpeg;base64,{image_base64} }, }, ], } ], ) return response.choices[0].message.content or def parse_result(self, raw: str) - dict: 从模型输出中提取JSON。 模型可能返回纯JSON也可能把JSON包裹在markdown代码块里。 cleaned raw.strip() code_block re.search(r(?:json)?\s*(.*?), cleaned, re.S) if code_block: cleaned code_block.group(1).strip() try: data json.loads(cleaned) except json.JSONDecodeError: start cleaned.find({) end cleaned.rfind(}) if start 0 and end start: try: data json.loads(cleaned[start : end 1]) except json.JSONDecodeError: return {} else: return {} return data if isinstance(data, dict) else {}有几个实现细节要说明。temperature设置为0.2目的是让模型输出更稳定。分析图表不是创意写作温度过高会导致同一张图每次结果差异很大。parse_result里做了两层兜底。第一层处理markdown代码块包裹的JSON第二层提取第一个{到最后一个}之间的内容。这样做能覆盖大部分模型输出格式不稳定的问题但并不能覆盖全部后面排错部分还会细说。注意这个Provider只写了调用和解析没有做重试。生产环境还需要补上指数退避重试否则某个模型限流时整个任务会失败。4.3 并行调度多个模型之间没有依赖关系天然适合并发调用。这里采用ThreadPoolExecutor因为API调用是IO密集型任务等待网络响应时线程会切换不会浪费CPU。numpy等计算密集型任务才需要进程池。在main.py中实现调度函数from concurrent.futures import ThreadPoolExecutor, as_completed def build_prompt() - str: return ( 你是一位量化交易分析师。请仔细查看用户提供的K线图 完成以下分析并严格输出JSON不要输出多余文字。\n 字段定义\n trend: bullish/bearish/neutral/uncertain\n confidence: 0到1之间的小数\n key_levels: 数组每项包含price和type(support/resistance)\n risk_points: 风险点字符串数组\n action_suggestion: 交易建议\n analysis_detail: 分析理由\n ) def analyze_with_all_models(providers, image_base64, prompt, max_workers4): results {} with ThreadPoolExecutor(max_workersmax_workers) as executor: future_map { executor.submit(provider.analyze_chart, image_base64, prompt): provider for provider in providers } for future in as_completed(future_map, timeout90): provider future_map[future] try: raw future.result() results[provider.name] {raw: raw, error: None} except Exception as exc: results[provider.name] {raw: , error: str(exc)} return results这里要注意as_completed(future_map, timeout90)的timeout是整个等待的总超时不是每个任务的超时。所以配置里的timeout要同时考虑模型数量和模型本身耗时。单个模型失败时不要直接抛异常让整个程序退出。更合理的做法是把错误记录下来让其他模型的结果继续参与汇总。失败模型的卡片显示“当前不可用”用户和管理员都能看到。4.4 结构化结果解析与汇总解析完成之后进入汇总阶段。汇总模块要解决四个问题趋势判断不一致时怎么办。置信度如何合并。关键点位重复和冲突如何处理。汇总结果如何保留原始差异而不是只显示一个最终值。在core/aggregator.py中实现from collections import Counter from typing import Dict, List def aggregate_results(model_outputs: Dict[str, dict]) - dict: 汇总多个模型的解析结果。 Args: model_outputs: {model_name: parsed_dict} Returns: 汇总报告。 valid_results {name: data for name, data in model_outputs.items() if data} if not valid_results: return {error: no_valid_model_output} trend_counter Counter(data.get(trend, uncertain) for data in valid_results.values()) top_trend trend_counter.most_common(1)[0][0] trend_votes dict(trend_counter) confidences [ float(data.get(confidence, 0.0)) for data in valid_results.values() if isinstance(data.get(confidence, 0.0), (int, float)) ] avg_confidence sum(confidences) / len(confidences) if confidences else 0.0 all_levels [] for data in valid_results.values(): levels data.get(key_levels, []) if isinstance(levels, list): all_levels.extend(levels) merged_levels _dedupe_levels(all_levels) all_risks [] for data in valid_results.values(): risks data.get(risk_points, []) if isinstance(risks, list): all_risks.extend(risks) merged_risks _dedupe_text(all_risks) suggestions [ data.get(action_suggestion, ) for data in valid_results.values() if data.get(action_suggestion) ] return { trend_summary: top_trend, trend_votes: trend_votes, avg_confidence: round(avg_confidence, 2), key_levels: merged_levels, risk_points: merged_risks, action_suggestions: suggestions, model_count: len(valid_results), model_names: list(valid_results.keys()), } def _dedupe_levels(levels, tolerance0.01): seen_prices [] result [] for item in levels: if not isinstance(item, dict): continue try: price float(item.get(price)) except (TypeError, ValueError): continue if any(abs(price - p) tolerance for p in seen_prices): continue seen_prices.append(price) result.append({ price: price, type: item.get(type, support), }) return sorted(result, keylambda x: x[price]) def _dedupe_text(items): cleaned [] seen set() for item in items: if not isinstance(item, str): continue text item.strip() if text and text not in seen: seen.add(text) cleaned.append(text) return cleaned这段汇总逻辑采用了“多数趋势 平均置信度 点位去重 风险合并”的策略。关键决策是如果三个模型里两个看多、一个看空trend_summary显示看多但trend_votes里能看出存在分歧risk_points中也会包含看空模型的风险提示。这样汇总报告既给了结论又保留了对立观点。_dedupe_levels里的tolerance0.01表示价格相差小于0.01的点位视为同一个位置。实际项目要根据交易品种调整这个值股票价格可能在几十元到几百元0.01太严格数字货币价格可能在几万美元0.01又太严格。更合理的做法是按价格比例去重比如abs(price - p) / p 0.001。5. 运行验证与结果分析5.1 用同一张图跑一次多模型分析为了本地验证先写一个生成样例K线图的脚本。sample/kline_sample.pyimport matplotlib.pyplot as plt import mplfinance as mpf import pandas as pd import numpy as np np.random.seed(42) n 120 dates pd.date_range(start2025-01-01, periodsn, freqD) price 100 np.cumsum(np.random.randn(n) * 1.2) noise np.random.randn(n) * 0.3 df pd.DataFrame({ Open: price noise, High: price np.abs(np.random.randn(n)) * 2 1, Low: price - np.abs(np.random.randn(n)) * 2 - 1, Close: price np.random.randn(n) * 0.5, }, indexdates) mpf.plot(df, typecandle, volumeTrue, stylecharles, savefigsample/kline_sample.png)运行python sample/kline_sample.py生成sample/kline_sample.png后写一个完整的入口main.pyimport os import sys import yaml from core.aggregator import aggregate_results from core.image_utils import preprocess_image from providers.base import BaseProvider from providers.openai_compatible import OpenAICompatibleProvider def load_providers(config): providers [] for item in config[models]: provider OpenAICompatibleProvider( nameitem[name], api_key_envitem[api_key_env], base_urlitem[base_url], modelitem[model], timeoutitem.get(timeout, 60), ) providers.append(provider) return providers def main(image_path, config_pathconfig.yaml): with open(config_path, r, encodingutf-8) as f: config yaml.safe_load(f) providers load_providers(config) image_base64 preprocess_image( image_path, max_sizeconfig[models][0].get(max_image_size, 1024), ) prompt build_prompt() raw_results analyze_with_all_models( providers, image_base64, prompt, max_workersconfig.get(parallel, {}).get(max_workers, 4), ) parsed_results {} for name, item in raw_results.items(): if item[error]: print(f[{name}] 调用失败: {item[error]}) continue parsed providers_map[name].parse_result(item[raw]) if not parsed: print(f[{name}] 解析失败原始输出: {item[raw][:200]}) parsed_results[name] parsed summary aggregate_results(parsed_results) import json print(json.dumps(summary, ensure_asciiFalse, indent2)) if __name__ __main__: image_path sys.argv[1] if len(sys.argv) 1 else sample/kline_sample.png main(image_path)运行前设置环境变量。export DASHSCOPE_API_KEY你的通义千问API Key export ZHIPU_API_KEY你的智谱API Key python main.py sample/kline_sample.png正常情况下控制台会输出类似这样的汇总结果{ trend_summary: bullish, trend_votes: { bullish: 2, bearish: 1 }, avg_confidence: 0.68, key_levels: [ {price: 96.5, type: support}, {price: 99.2, type: support}, {price: 103.8, type: resistance} ], risk_points: [ 价格接近前期高点存在回调压力, 成交量未明显放大突破有效性待确认 ], action_suggestions: [ 偏多持有跌破99元支撑位再考虑减仓, 观望等待放量突破103元后追入, 不建议在当前位置重仓等待回踩确认 ], model_count: 3, model_names: [qwen-vl-plus, glm-4v-plus] }注意这个输出结构是示例格式实际内容取决于你接入的模型和当前图表走势。如果某个模型没有配API Key它会出现在失败名单里其他模型的结果仍会输出。5.2 怎么判断多模型结果是否可信多模型汇总输出后不能只看trend_summary。我建议按三条规则判断第一条规则没有共识的结论不要用。如果三个模型分别给出看多、看空、中性说明图表信息存在分歧这时候trend_summary只是“少数服从多数”没有实际决策价值。第二条规则置信度只能相对比较。模型说confidence0.9不代表真实概率是90%。但同一个模型对两张图分别给出0.9和0.5时0.9的那张图通常更值得关注。第三条规则空跑测试不可少。拿一张没有明显趋势的横盘图去测如果多个模型仍然编造出“强支撑位”“突破压力位”说明提示词太容易被带偏或者模型本身幻觉严重。应该让模型在不确定性高时输出uncertain而不是硬给一个多空判断。5.3 一致性指标为了衡量“多个AI是否真的在看同一张图”可以引入简单的一致性指标。指标计算方式含义多数一致率最多trend票数 / 有效模型数判断结论的集中度置信度标准差所有模型confidence的标准差模型对图表把握的分歧程度关键点位重合率重合点位数量 / 全部点位数量模型识别关键位置的稳定性风险点覆盖率风险点总数 / 有效模型数风险信息是否被充分表达这些指标不需要写得很复杂。如果项目刚起步只需要输出trend_votes和confidence字段就能人工判断多模型结果是否可信。指标本身是工具目的是让“模型之间差异大”这件事变得可见。6. 常见问题与排查路径6.1 API调用失败类问题现象一调用模型时返回401 Unauthorized。可能原因api_key_env指定的环境变量不存在或者Key本身无效。检查方式echo $DASHSCOPE_API_KEY如果输出为空说明环境变量没设置。如果设置了仍报401说明Key不正确或已被禁用。现象二返回400 Bad Request。常见原因有三种模型名拼错比如把qwen-vl-plus写成qwen-vl。图片base64格式错误缺少data:image/jpeg;base64,前缀。图片分辨率过大超出服务端限额。检查方式打印实际请求中的base_url、model、api_key_env。确认图片编码后字符串以/9j/开头这是JPEG的base64特征前缀。把max_image_size降到1024重试。现象三请求返回429 Too Many Requests。原因触发限流。可能因为并发数过高或者免费额度耗尽。处理建议在config.yaml中把max_workers降低到2或者给Provider增加重试逻辑。现象四请求超时或者长时间无响应。原因视觉模型推理慢60秒不一定够。建议先单独跑一个模型统计实际耗时再决定timeout设置。生产环境timeout要留出1.5倍余量。6.2 输出解析失败类问题现象一模型返回了大量文字但parse_result解析成空字典。原因提示词要求JSON但模型输出时在前面加了解释性文字或吐出了多个JSON对象。处理方式检查原始输出确认是否只有一个JSON对象。如果模型总是加前缀在提示词最后追加“只输出JSON对象不要解释”。现象二JSON字段缺失。原因模型返回了JSON但只包含analysis_detail没有trend和confidence。处理方式在汇总层做字段补齐缺少的字段设为默认值并在报告中标出“某模型字段不完整”。现象三trend字段出现枚举之外的字符串比如likely_bullish。原因模型对枚举定义理解不到位。处理方式在提示词里增加一个字段约束说明trend字段只能取这四个值之一bullish、bearish、neutral、uncertain。 如果趋势不明确必须输出uncertain不能编造其他值。6.3 分析质量异常类问题现象一模型把K线图里的水印、logo看成价格走势。原因用户上传的截图带了平台水印水印线条和K线重叠。处理方式在预处理阶段对图片顶部和底部做固定比例裁剪去掉常见的水印区域。如果水印在图中需要提示用户上传不带水印的截图。现象二模型把历史K线数据解读成未来预测。原因提示词没有明确说明“这是历史数据”。模型不知道当前时间点容易把最后一根K线之后的走势脑补出来。处理方式在提示词最前面增加一句用户提供的是历史K线图不是实时行情。请基于图中已有信息分析不要预测图中未显示的走势。现象三不同模型给出完全相反的结论汇总结果不稳定。原因这不一定是bug。模型训练数据、视觉偏好、对技术指标的理解不同分歧是正常的。处理方式不要通过修改模型来强行消除分歧。汇总报告里保留trend_votes把分歧作为风险提示。6.4 排错顺序清单遇到问题时按以下顺序排查能省掉大量无意义的日志翻找排查顺序检查内容工具或方法1输入图片路径是否存在是否可读ls -l、PIL打开2图片预处理是否成功base64是否非空打印编码前50字符3环境变量是否设置正确echo $变量名4base_url和model名是否有效用curl或官方示例验证5单模型请求是否超时单独跑一次并计时6原始输出格式是否符合预期打印未解析的原始文本7汇总层字段是否存在类型错误检查trend是否为字符串confidence是否可转float这个清单适用于大多数“多模型看一张图”场景。核心思路是从上游到下游逐段验证输入有问题就先解决输入不要先去猜模型是不是疯了。7. 生产化落地要点7.1 学习环境与生产环境的差异本地能跑通和线上能稳定运行是两回事。区别集中在资源、容错和可观测性上。维度学习环境生产环境API Key本地环境变量密钥管理系统或环境注入失败处理直接打印异常重试、降级、告警缓存无同一图片hash结果缓存日志不记录记录prompt、输出、耗时、token数限流不关注需要配额管理和熔断审计不关注每笔分析可追溯并发单机低并发多实例横向扩展图片存储本地临时文件对象存储带生命周期生产环境还要考虑一件事图片上传后不能无限制保留。K线截图本质是行情数据不同数据源可能有版权要求。建议图片上传后只保留分析报告图片本身在24小时内删除。7.2 生产环境还要补什么第一结果缓存。同一张图在几分钟内被多次分析结果基本一样。可以用图片的SHA256值作为缓存key命中缓存就直接返回历史结论减少模型调用成本。import hashlib def image_hash(image_bytes: bytes) - str: return hashlib.sha256(image_bytes).hexdigest()第二模型调用降级。三个模型里有一个挂了不能导致整个服务不可用。汇总层要支持“缺失模型”场景在报告中标注“本次由2/3模型参与分析”。第三可观测性。至少要记录的是哪个模型、耗时多少、token消耗多少、输出是否解析成功、单次调用总成本。没有这些数据后续优化模型选型和并发参数都无从下手。第四提示词版本管理。提示词是核心资产改一句话可能显著影响分析结果。建议把提示词存成模板文件随代码发布而不是硬编码在源码里。线上出问题时能快速回滚到上一个提示词版本。7.3 多模型决策融合的几种做法汇总策略不能一刀切要按使用场景选择。融合方式实现难度优点缺点适用场景简单投票低直观、易调试每个模型权重相同忽略了置信度差异分析报告展示置信度加权中高置信度模型影响更大模型置信度可能虚高多空倾向评估多数趋势 分歧提示中保留不同观点结论不够干脆辅助决策模型结果 规则过滤高叠加技术指标和风控规则模型输出成为中间特征依赖解析质量策略信号生成量化交易场景里我比较推荐“多数趋势 分歧提示 规则后置”的组合。先让多个AI输出观点再用程序化的技术指标验证模型的判断。比如模型看多同时5日均线上穿20日均线信号置信度就可以提高如果模型看多但RSI进入超买区报告里应该追加一条警示。7.4 合规与风险提示不建议越过哪些边界多AI看一张图这个功能产品价值确实存在但有几条边界不建议越过。第一不要在界面里出现“AI建议直接买入”“AI预测必涨”这类表述。模型的输出本质是概率性推测不是事实。所有结论都必须标注“仅供研究参考不构成投资建议”。第二不要把模型的置信度渲染成收益预测。confidence0.9只能说明模型对自己判断的确定性高不能翻译成“90%概率赚钱”。第三API Key不要提交到git仓库。建议在.gitignore中加入.env并使用环境变量或密钥管理服务。第四如果产品面向公众用户需要在显著位置提示用户AI视觉分析会受到图片清晰度、指标设置、数据范围等因素影响结论可能不够准确。第五回测不能说明未来。用历史K线图测出“模型判断很准”之后不要直接上实盘。视觉模型容易记住训练数据里的形态但历史规律会随市场结构变化失效。8. 从演示到可用最该盯住哪几件事整套代码写完能跑通多模型分析之后最值得花时间的不是继续增加模型数量而是做三件事。第一件事建立自己的评估集。准备20到30张不同类型的K线图包含明显的上升趋势、下降趋势、横盘震荡、长上影线、放量突破等场景记录每个模型的输出。不要只看最终结论还要看关键点位、风险点、分析理由是不是符合常识。没有评估集你永远不知道模型什么时候在胡说。第二件事锁死输出协议。多模型接入最大的风险不是模型能力不够而是输出格式不稳定。一个模型返回了完整JSON另一个模型在JSON前后加了大量解释汇总层就被迫做各种容错。应该让所有模型共用同一份严格提示词模板并用测试用例覆盖“解析失败”“字段缺失”等异常分支。第三件事先在辅助场景运行。把多AI分析结果放到日报、周报、研究草稿这类“有人工复核”的功能里观察一段时间再考虑是否把结论接入自动提醒或策略信号。直接接入交易执行系统现阶段风险远大于收益。这套多模型看图的实现本质上是一个很典型的“模型编排 结构化输出 聚合决策”工程问题。即使以后换成更新的模型只要保持统一接口、统一输出协议、统一汇总逻辑迁移成本都不会太大。对想入门的开发者来说这也是一个把大模型能力从“聊天玩具”往“可计算工具”推进的合适练习项目。 多模型AI协同看K线图：从原理到最小可实现系统
AI量化交易最近讨论最多的方向已经从“用AI写策略”变成了“让AI直接读图并给出交易分析”。所谓多个AI同时看一张图是指把同一张K线图、分时图或技术指标图同时发送给多个大语言模型让每个模型独立输出趋势判断、关键点位和风险提示再由系统汇总成一份综合结论。这种做法的价值不在于某一个模型有多强而在于多个模型在同一张图上互相印证、彼此纠偏。标题中提到的孙越AI正是这类多模型量化分析工具的一个产品化形态。下面用一套最小可运行实现拆解它的核心链路图表预处理、多模型并行调用、结果解析和综合汇总并给出本地验证和生产落地时最容易踩的坑。1. 多AI看一张图先想清楚它解决什么问题1.1 单模型分析量化图表的局限性传统量化分析有两种常见做法。一种靠人工盯盘看K线形态、均线位置、成交量变化然后凭经验判断。另一种靠规则引擎把“MACD金叉”“突破20日均线”这类条件写成程序满足条件就触发信号。前者的问题是主观、容易疲劳同一个人在不同时间看同一张图结论都可能不一样。后者的问题更隐蔽规则引擎只能识别写死的形态无法理解形态背后的市场氛围也没有办法解释“为什么这里可能是支撑位”。大模型出现后很多人尝试让模型直接看K线截图用自然语言输出分析。效果比预想好但单独用一个模型做这件事会暴露出四类问题视觉理解偏差。模型对K线实体、影线、均线交叉、成交量放量的理解不一定准确经常会把长上影线看成强势突破。上下文遗漏。K线图信息密度高模型在长上下文里可能漏掉关键位置比如前期高点、密集成交区。过度自信。即使模型判断错了它也会用很确定的语气输出“大概率上涨”缺少自我怀疑。风格偏置。同一个任务有的模型习惯看多有的模型习惯看空单模型会导致结论体系性偏移。所以“多个AI同时看一张图”不是炫技而是用模型组合来对冲单模型的不确定性。1.2 多模型并行分析的产品形态把多个AI同时看一张图拆开看本质是“多智能体评审”。系统把同一张图同时分发给多个大模型每个模型独立输出结构化报告再由汇总模块生成综合结论。产品端呈现通常是这样用户上传一张K线截图或分时图。系统显示多个模型的分析卡片每个卡片包含趋势判断、置信度、关键点位、风险点。顶部显示汇总结论标出哪些模型观点一致哪些模型存在分歧。这里的核心并不在于“谁对谁错”而在于两点共识部分是否稳定分歧部分能否提示风险。如果三个模型都看多结论可信度就比单个模型看多高得多。如果两个看多、一个看空系统应该把这个分歧明确提示出来而不是强行取平均。这里的预期收益不是“准确率提升几个点”而是风险表达更完整单模型漏掉的风险另一个模型可能会补上。1.3 这个功能适合用在什么场景并不是所有量化场景都适合让AI看图表。适合的是那些“需要快速生成分析草稿、需要多视角解读、需要人工复核”的场景。场景输入输出人工介入程度个人看盘辅助K线截图多模型分析报告中用户自己判断策略研究历史K线片段形态识别和特征汇总高研究员需复核行情日报自动生成当日截图多空观点汇总中编辑审核后发布AI教学演示任意图表不同模型的解读对比低展示差异为主自动下单实时行情图买卖信号极高不建议完全自动化最后一行尤其重要。现阶段让多个AI看一张图适合作为“分析助手”不适合直接接管交易。原因很简单模型没有实时行情权限视觉推理本身存在误差而且它无法感知盘口深度、资金流向、突发消息等非图表信息。2. 系统设计与技术拆分2.1 整体处理链路图表输入、模型分发、结果汇总多AI看一张图技术链路可以拆成五步接收用户上传的图片。对图片做统一预处理裁剪多余区域、缩放分辨率、转base64编码。把同一张图片和同一份提示词并发发送给多个模型。每个模型返回自然语言或JSON文本解析成统一结构。汇总模块对多个结构化结果做投票、合并和风险提示最后生成报告。这条链路最容易被忽略的环节是第一步和第二步。很多实现把用户上传的原图直接发给模型图片分辨率过高会导致token消耗暴涨过低又会导致模型看不清K线形态。后面会单独讲预处理参数。分发环节要注意“并发”和“超时”。多模型串行调用总耗时等于所有模型耗时相加体验会很差。正确做法是并发调用同时给每个模型设置独立的超时时间避免某个模型卡住拖垮整个任务。汇总环节不是简单拼接。如果只是把多个模型的文字结果堆在一起用户还得自己对比体验和看多个网页没有区别。汇总必须解决“观点是否一致”“分歧在哪里”“最终建议是什么”这三个问题。2.2 模型供应商层用兼容接口屏蔽差异不同模型厂商的协议不统一如果为每个模型写一套单独调用代码维护成本会很高。实践中建议统一使用OpenAI兼容协议。当前主流模型的API服务多数都提供OpenAI兼容的endpoint只需要把base_url改成对应厂商的地址同时替换api_key和model名称即可。在设计模型供应商层时要抽象出一个基础接口方法作用输入输出analyze_chart核心分析入口图片base64、文本提示词模型原始输出文本parse_result解析模型输出原始文本结构化字典health_check连通性检测无是否可用这样做的好处是后续接入新模型时不需要改动上层调度和汇总逻辑只需要新增一个配置项或一个Provider类。接入模型时需要确认这类信息先看模型是否支持图像输入vision能力再看接口地址和鉴权方式最后确认单张图片的token上限。写配置时不要照搬网上的base_url要以对应服务商当前文档为准。2.3 提示词与输出协议设计多模型分析成败的另一个关键点是输出协议。如果让模型自由发挥返回的可能是长段落也可能是列表汇总程序很难解析。建议在提示词中强制规定JSON输出结构{ trend: bullish, confidence: 0.7, key_levels: [ {price: 100.5, type: support}, {price: 105.0, type: resistance} ], risk_points: [ 成交量萎缩上涨动力不足 ], action_suggestion: 短期持有跌破支撑位减仓, analysis_detail: 均线多头排列但MACD红柱缩短 }各字段含义如下trend多空判断枚举值bullish、bearish、neutral、uncertain。confidence模型对自己判断的置信度0到1之间。key_levels关键支撑和压力位列表。risk_points风险点列表字符串数组。action_suggestion交易动作建议保留自然语言便于展示。analysis_detail分析理由用于生成详细报告。统一结构之后无论模型来自哪家厂商汇总模块都只需要处理同一套字典格式。3. 环境准备与依赖3.1 版本和依赖清单下面的实现基于Python 3.10本地开发可以直接用venv隔离环境。创建虚拟环境python3 -m venv venv source venv/bin/activate安装依赖前先确认pip版本和Python版本一致。python --version pip --version然后在项目根目录创建requirements.txtopenai1.35.0 Pillow10.0.0 requests2.31.0 PyYAML6.0.1 matplotlib3.8.0逐项说明用途依赖用途openai调用OpenAI兼容接口新版客户端统一用chat.completionsPillow图片缩放、格式转换、base64编码requests健康检查或调试接口PyYAML读取模型配置文件matplotlib生成样例K线图方便本地验证安装后确认版本特别要注意openai库版本。1.0前后API差异很大旧项目的写法是openai.ChatCompletion.create新版写法是client.chat.completions.create。3.2 项目目录结构建议按模块拆分方便后续扩展multi_ai_chart/ ├── config.yaml ├── requirements.txt ├── main.py ├── providers/ │ ├── __init__.py │ ├── base.py │ └── openai_compatible.py ├── core/ │ ├── __init__.py │ ├── image_utils.py │ └── aggregator.py └── sample/ └── kline_sample.py各文件职责文件职责main.py入口读取配置、加载图片、调度模型、汇总输出providers/base.py模型供应商抽象基类providers/openai_compatible.py兼容接口实现core/image_utils.py图片预处理与base64编码core/aggregator.py多模型结果汇总sample/kline_sample.py生成样例K线图config.yaml模型列表和参数配置这种结构在本地开发时够用上线时可以把providers和core拆成独立服务。3.3 配置文件在项目根目录创建config.yamlmodels: - name: qwen-vl-plus display_name: Qwen-VL base_url: https://dashscope.aliyuncs.com/compatible-mode/v1 api_key_env: DASHSCOPE_API_KEY model: qwen-vl-plus max_image_size: 1024 timeout: 60 - name: glm-4v-plus display_name: GLM-4V base_url: https://open.bigmodel.cn/api/paas/v4 api_key_env: ZHIPU_API_KEY model: glm-4v-plus max_image_size: 1024 timeout: 60 parallel: max_workers: 4 timeout: 90 output: language: zh-CN这里的关键参数说明参数含义推荐值影响max_image_size图片最长边像素1024过大消耗token过小模型看不清timeout单个模型请求超时60秒视觉模型推理较慢不能设太短max_workers并发线程数模型数量并发太高容易触发限流api_key_envAPI Key所在环境变量名自定义不要把明文Key写进配置注意不同厂商的base_url会变化落地前应以对应服务商当前文档为准。上述地址是常见的兼容接口格式只用于示例。4. 核心代码实现4.1 图表预处理裁剪、缩放、转Base64图片预处理是很容易被忽略的一步。用户上传的截图可能是整个屏幕包含广告栏、标题、水印这些噪声会干扰模型识别也会浪费token。建议处理顺序是打开图片。可选裁剪去掉上下边缘非图表区域。按最长边缩放到max_image_size。转成RGB模式避免RGBA图片带透明通道导致编码异常。压缩为JPEG格式质量设为90。转base64字符串。实现core/image_utils.py中的核心函数import base64 from io import BytesIO from PIL import Image def preprocess_image( image_path: str, max_size: int 1024, quality: int 90, ) - str: 读取图片缩放并编码为base64字符串。 Args: image_path: 本地图片路径。 max_size: 最长边像素超过则等比缩放。 quality: JPEG压缩质量0-100。 Returns: base64编码的JPEG图片。 with Image.open(image_path) as img: img img.convert(RGB) w, h img.size max_side max(w, h) if max_side max_size: ratio max_size / max_side new_w int(w * ratio) new_h int(h * ratio) img img.resize((new_w, new_h), Image.LANCZOS) buffer BytesIO() img.save(buffer, formatJPEG, qualityquality) encoded base64.b64encode(buffer.getvalue()).decode(utf-8) return encoded这段代码有三个关键点。第一缩放而不是裁剪。等比缩放不会改变K线形状模型对实体、影线、均线方向仍然能识别。第二转RGB再保存JPEG。PNG图片带透明通道时直接编码可能导致部分服务端解析异常转RGB后更稳。第三JPEG质量90能平衡清晰度和体积。K线图是矢量特征明显的图形质量降到85以下时细小的均线标记可能变糊。4.2 多模型客户端封装在providers/base.py中定义抽象基类from abc import ABC, abstractmethod from typing import Dict class BaseProvider(ABC): 模型供应商抽象基类。 def __init__(self, name: str): self.name name abstractmethod def analyze_chart(self, image_base64: str, prompt: str) - str: 输入图片和提示词返回模型原始文本。 raise NotImplementedError abstractmethod def parse_result(self, raw: str) - Dict: 把模型返回的原始文本解析成统一结构。 raise NotImplementedError在providers/openai_compatible.py中实现OpenAI兼容协议import json import os import re from openai import OpenAI from providers.base import BaseProvider class OpenAICompatibleProvider(BaseProvider): OpenAI兼容协议模型供应商。 def __init__( self, name: str, api_key_env: str, base_url: str, model: str, timeout: int 60, temperature: float 0.2, ): super().__init__(name) api_key os.getenv(api_key_env) if not api_key: raise ValueError(f环境变量 {api_key_env} 未设置) self.client OpenAI(api_keyapi_key, base_urlbase_url, timeouttimeout) self.model model self.temperature temperature def analyze_chart(self, image_base64: str, prompt: str) - str: response self.client.chat.completions.create( modelself.model, temperatureself.temperature, messages[ { role: user, content: [ {type: text, text: prompt}, { type: image_url, image_url: { url: fdata:image/jpeg;base64,{image_base64} }, }, ], } ], ) return response.choices[0].message.content or def parse_result(self, raw: str) - dict: 从模型输出中提取JSON。 模型可能返回纯JSON也可能把JSON包裹在markdown代码块里。 cleaned raw.strip() code_block re.search(r(?:json)?\s*(.*?), cleaned, re.S) if code_block: cleaned code_block.group(1).strip() try: data json.loads(cleaned) except json.JSONDecodeError: start cleaned.find({) end cleaned.rfind(}) if start 0 and end start: try: data json.loads(cleaned[start : end 1]) except json.JSONDecodeError: return {} else: return {} return data if isinstance(data, dict) else {}有几个实现细节要说明。temperature设置为0.2目的是让模型输出更稳定。分析图表不是创意写作温度过高会导致同一张图每次结果差异很大。parse_result里做了两层兜底。第一层处理markdown代码块包裹的JSON第二层提取第一个{到最后一个}之间的内容。这样做能覆盖大部分模型输出格式不稳定的问题但并不能覆盖全部后面排错部分还会细说。注意这个Provider只写了调用和解析没有做重试。生产环境还需要补上指数退避重试否则某个模型限流时整个任务会失败。4.3 并行调度多个模型之间没有依赖关系天然适合并发调用。这里采用ThreadPoolExecutor因为API调用是IO密集型任务等待网络响应时线程会切换不会浪费CPU。numpy等计算密集型任务才需要进程池。在main.py中实现调度函数from concurrent.futures import ThreadPoolExecutor, as_completed def build_prompt() - str: return ( 你是一位量化交易分析师。请仔细查看用户提供的K线图 完成以下分析并严格输出JSON不要输出多余文字。\n 字段定义\n trend: bullish/bearish/neutral/uncertain\n confidence: 0到1之间的小数\n key_levels: 数组每项包含price和type(support/resistance)\n risk_points: 风险点字符串数组\n action_suggestion: 交易建议\n analysis_detail: 分析理由\n ) def analyze_with_all_models(providers, image_base64, prompt, max_workers4): results {} with ThreadPoolExecutor(max_workersmax_workers) as executor: future_map { executor.submit(provider.analyze_chart, image_base64, prompt): provider for provider in providers } for future in as_completed(future_map, timeout90): provider future_map[future] try: raw future.result() results[provider.name] {raw: raw, error: None} except Exception as exc: results[provider.name] {raw: , error: str(exc)} return results这里要注意as_completed(future_map, timeout90)的timeout是整个等待的总超时不是每个任务的超时。所以配置里的timeout要同时考虑模型数量和模型本身耗时。单个模型失败时不要直接抛异常让整个程序退出。更合理的做法是把错误记录下来让其他模型的结果继续参与汇总。失败模型的卡片显示“当前不可用”用户和管理员都能看到。4.4 结构化结果解析与汇总解析完成之后进入汇总阶段。汇总模块要解决四个问题趋势判断不一致时怎么办。置信度如何合并。关键点位重复和冲突如何处理。汇总结果如何保留原始差异而不是只显示一个最终值。在core/aggregator.py中实现from collections import Counter from typing import Dict, List def aggregate_results(model_outputs: Dict[str, dict]) - dict: 汇总多个模型的解析结果。 Args: model_outputs: {model_name: parsed_dict} Returns: 汇总报告。 valid_results {name: data for name, data in model_outputs.items() if data} if not valid_results: return {error: no_valid_model_output} trend_counter Counter(data.get(trend, uncertain) for data in valid_results.values()) top_trend trend_counter.most_common(1)[0][0] trend_votes dict(trend_counter) confidences [ float(data.get(confidence, 0.0)) for data in valid_results.values() if isinstance(data.get(confidence, 0.0), (int, float)) ] avg_confidence sum(confidences) / len(confidences) if confidences else 0.0 all_levels [] for data in valid_results.values(): levels data.get(key_levels, []) if isinstance(levels, list): all_levels.extend(levels) merged_levels _dedupe_levels(all_levels) all_risks [] for data in valid_results.values(): risks data.get(risk_points, []) if isinstance(risks, list): all_risks.extend(risks) merged_risks _dedupe_text(all_risks) suggestions [ data.get(action_suggestion, ) for data in valid_results.values() if data.get(action_suggestion) ] return { trend_summary: top_trend, trend_votes: trend_votes, avg_confidence: round(avg_confidence, 2), key_levels: merged_levels, risk_points: merged_risks, action_suggestions: suggestions, model_count: len(valid_results), model_names: list(valid_results.keys()), } def _dedupe_levels(levels, tolerance0.01): seen_prices [] result [] for item in levels: if not isinstance(item, dict): continue try: price float(item.get(price)) except (TypeError, ValueError): continue if any(abs(price - p) tolerance for p in seen_prices): continue seen_prices.append(price) result.append({ price: price, type: item.get(type, support), }) return sorted(result, keylambda x: x[price]) def _dedupe_text(items): cleaned [] seen set() for item in items: if not isinstance(item, str): continue text item.strip() if text and text not in seen: seen.add(text) cleaned.append(text) return cleaned这段汇总逻辑采用了“多数趋势 平均置信度 点位去重 风险合并”的策略。关键决策是如果三个模型里两个看多、一个看空trend_summary显示看多但trend_votes里能看出存在分歧risk_points中也会包含看空模型的风险提示。这样汇总报告既给了结论又保留了对立观点。_dedupe_levels里的tolerance0.01表示价格相差小于0.01的点位视为同一个位置。实际项目要根据交易品种调整这个值股票价格可能在几十元到几百元0.01太严格数字货币价格可能在几万美元0.01又太严格。更合理的做法是按价格比例去重比如abs(price - p) / p 0.001。5. 运行验证与结果分析5.1 用同一张图跑一次多模型分析为了本地验证先写一个生成样例K线图的脚本。sample/kline_sample.pyimport matplotlib.pyplot as plt import mplfinance as mpf import pandas as pd import numpy as np np.random.seed(42) n 120 dates pd.date_range(start2025-01-01, periodsn, freqD) price 100 np.cumsum(np.random.randn(n) * 1.2) noise np.random.randn(n) * 0.3 df pd.DataFrame({ Open: price noise, High: price np.abs(np.random.randn(n)) * 2 1, Low: price - np.abs(np.random.randn(n)) * 2 - 1, Close: price np.random.randn(n) * 0.5, }, indexdates) mpf.plot(df, typecandle, volumeTrue, stylecharles, savefigsample/kline_sample.png)运行python sample/kline_sample.py生成sample/kline_sample.png后写一个完整的入口main.pyimport os import sys import yaml from core.aggregator import aggregate_results from core.image_utils import preprocess_image from providers.base import BaseProvider from providers.openai_compatible import OpenAICompatibleProvider def load_providers(config): providers [] for item in config[models]: provider OpenAICompatibleProvider( nameitem[name], api_key_envitem[api_key_env], base_urlitem[base_url], modelitem[model], timeoutitem.get(timeout, 60), ) providers.append(provider) return providers def main(image_path, config_pathconfig.yaml): with open(config_path, r, encodingutf-8) as f: config yaml.safe_load(f) providers load_providers(config) image_base64 preprocess_image( image_path, max_sizeconfig[models][0].get(max_image_size, 1024), ) prompt build_prompt() raw_results analyze_with_all_models( providers, image_base64, prompt, max_workersconfig.get(parallel, {}).get(max_workers, 4), ) parsed_results {} for name, item in raw_results.items(): if item[error]: print(f[{name}] 调用失败: {item[error]}) continue parsed providers_map[name].parse_result(item[raw]) if not parsed: print(f[{name}] 解析失败原始输出: {item[raw][:200]}) parsed_results[name] parsed summary aggregate_results(parsed_results) import json print(json.dumps(summary, ensure_asciiFalse, indent2)) if __name__ __main__: image_path sys.argv[1] if len(sys.argv) 1 else sample/kline_sample.png main(image_path)运行前设置环境变量。export DASHSCOPE_API_KEY你的通义千问API Key export ZHIPU_API_KEY你的智谱API Key python main.py sample/kline_sample.png正常情况下控制台会输出类似这样的汇总结果{ trend_summary: bullish, trend_votes: { bullish: 2, bearish: 1 }, avg_confidence: 0.68, key_levels: [ {price: 96.5, type: support}, {price: 99.2, type: support}, {price: 103.8, type: resistance} ], risk_points: [ 价格接近前期高点存在回调压力, 成交量未明显放大突破有效性待确认 ], action_suggestions: [ 偏多持有跌破99元支撑位再考虑减仓, 观望等待放量突破103元后追入, 不建议在当前位置重仓等待回踩确认 ], model_count: 3, model_names: [qwen-vl-plus, glm-4v-plus] }注意这个输出结构是示例格式实际内容取决于你接入的模型和当前图表走势。如果某个模型没有配API Key它会出现在失败名单里其他模型的结果仍会输出。5.2 怎么判断多模型结果是否可信多模型汇总输出后不能只看trend_summary。我建议按三条规则判断第一条规则没有共识的结论不要用。如果三个模型分别给出看多、看空、中性说明图表信息存在分歧这时候trend_summary只是“少数服从多数”没有实际决策价值。第二条规则置信度只能相对比较。模型说confidence0.9不代表真实概率是90%。但同一个模型对两张图分别给出0.9和0.5时0.9的那张图通常更值得关注。第三条规则空跑测试不可少。拿一张没有明显趋势的横盘图去测如果多个模型仍然编造出“强支撑位”“突破压力位”说明提示词太容易被带偏或者模型本身幻觉严重。应该让模型在不确定性高时输出uncertain而不是硬给一个多空判断。5.3 一致性指标为了衡量“多个AI是否真的在看同一张图”可以引入简单的一致性指标。指标计算方式含义多数一致率最多trend票数 / 有效模型数判断结论的集中度置信度标准差所有模型confidence的标准差模型对图表把握的分歧程度关键点位重合率重合点位数量 / 全部点位数量模型识别关键位置的稳定性风险点覆盖率风险点总数 / 有效模型数风险信息是否被充分表达这些指标不需要写得很复杂。如果项目刚起步只需要输出trend_votes和confidence字段就能人工判断多模型结果是否可信。指标本身是工具目的是让“模型之间差异大”这件事变得可见。6. 常见问题与排查路径6.1 API调用失败类问题现象一调用模型时返回401 Unauthorized。可能原因api_key_env指定的环境变量不存在或者Key本身无效。检查方式echo $DASHSCOPE_API_KEY如果输出为空说明环境变量没设置。如果设置了仍报401说明Key不正确或已被禁用。现象二返回400 Bad Request。常见原因有三种模型名拼错比如把qwen-vl-plus写成qwen-vl。图片base64格式错误缺少data:image/jpeg;base64,前缀。图片分辨率过大超出服务端限额。检查方式打印实际请求中的base_url、model、api_key_env。确认图片编码后字符串以/9j/开头这是JPEG的base64特征前缀。把max_image_size降到1024重试。现象三请求返回429 Too Many Requests。原因触发限流。可能因为并发数过高或者免费额度耗尽。处理建议在config.yaml中把max_workers降低到2或者给Provider增加重试逻辑。现象四请求超时或者长时间无响应。原因视觉模型推理慢60秒不一定够。建议先单独跑一个模型统计实际耗时再决定timeout设置。生产环境timeout要留出1.5倍余量。6.2 输出解析失败类问题现象一模型返回了大量文字但parse_result解析成空字典。原因提示词要求JSON但模型输出时在前面加了解释性文字或吐出了多个JSON对象。处理方式检查原始输出确认是否只有一个JSON对象。如果模型总是加前缀在提示词最后追加“只输出JSON对象不要解释”。现象二JSON字段缺失。原因模型返回了JSON但只包含analysis_detail没有trend和confidence。处理方式在汇总层做字段补齐缺少的字段设为默认值并在报告中标出“某模型字段不完整”。现象三trend字段出现枚举之外的字符串比如likely_bullish。原因模型对枚举定义理解不到位。处理方式在提示词里增加一个字段约束说明trend字段只能取这四个值之一bullish、bearish、neutral、uncertain。 如果趋势不明确必须输出uncertain不能编造其他值。6.3 分析质量异常类问题现象一模型把K线图里的水印、logo看成价格走势。原因用户上传的截图带了平台水印水印线条和K线重叠。处理方式在预处理阶段对图片顶部和底部做固定比例裁剪去掉常见的水印区域。如果水印在图中需要提示用户上传不带水印的截图。现象二模型把历史K线数据解读成未来预测。原因提示词没有明确说明“这是历史数据”。模型不知道当前时间点容易把最后一根K线之后的走势脑补出来。处理方式在提示词最前面增加一句用户提供的是历史K线图不是实时行情。请基于图中已有信息分析不要预测图中未显示的走势。现象三不同模型给出完全相反的结论汇总结果不稳定。原因这不一定是bug。模型训练数据、视觉偏好、对技术指标的理解不同分歧是正常的。处理方式不要通过修改模型来强行消除分歧。汇总报告里保留trend_votes把分歧作为风险提示。6.4 排错顺序清单遇到问题时按以下顺序排查能省掉大量无意义的日志翻找排查顺序检查内容工具或方法1输入图片路径是否存在是否可读ls -l、PIL打开2图片预处理是否成功base64是否非空打印编码前50字符3环境变量是否设置正确echo $变量名4base_url和model名是否有效用curl或官方示例验证5单模型请求是否超时单独跑一次并计时6原始输出格式是否符合预期打印未解析的原始文本7汇总层字段是否存在类型错误检查trend是否为字符串confidence是否可转float这个清单适用于大多数“多模型看一张图”场景。核心思路是从上游到下游逐段验证输入有问题就先解决输入不要先去猜模型是不是疯了。7. 生产化落地要点7.1 学习环境与生产环境的差异本地能跑通和线上能稳定运行是两回事。区别集中在资源、容错和可观测性上。维度学习环境生产环境API Key本地环境变量密钥管理系统或环境注入失败处理直接打印异常重试、降级、告警缓存无同一图片hash结果缓存日志不记录记录prompt、输出、耗时、token数限流不关注需要配额管理和熔断审计不关注每笔分析可追溯并发单机低并发多实例横向扩展图片存储本地临时文件对象存储带生命周期生产环境还要考虑一件事图片上传后不能无限制保留。K线截图本质是行情数据不同数据源可能有版权要求。建议图片上传后只保留分析报告图片本身在24小时内删除。7.2 生产环境还要补什么第一结果缓存。同一张图在几分钟内被多次分析结果基本一样。可以用图片的SHA256值作为缓存key命中缓存就直接返回历史结论减少模型调用成本。import hashlib def image_hash(image_bytes: bytes) - str: return hashlib.sha256(image_bytes).hexdigest()第二模型调用降级。三个模型里有一个挂了不能导致整个服务不可用。汇总层要支持“缺失模型”场景在报告中标注“本次由2/3模型参与分析”。第三可观测性。至少要记录的是哪个模型、耗时多少、token消耗多少、输出是否解析成功、单次调用总成本。没有这些数据后续优化模型选型和并发参数都无从下手。第四提示词版本管理。提示词是核心资产改一句话可能显著影响分析结果。建议把提示词存成模板文件随代码发布而不是硬编码在源码里。线上出问题时能快速回滚到上一个提示词版本。7.3 多模型决策融合的几种做法汇总策略不能一刀切要按使用场景选择。融合方式实现难度优点缺点适用场景简单投票低直观、易调试每个模型权重相同忽略了置信度差异分析报告展示置信度加权中高置信度模型影响更大模型置信度可能虚高多空倾向评估多数趋势 分歧提示中保留不同观点结论不够干脆辅助决策模型结果 规则过滤高叠加技术指标和风控规则模型输出成为中间特征依赖解析质量策略信号生成量化交易场景里我比较推荐“多数趋势 分歧提示 规则后置”的组合。先让多个AI输出观点再用程序化的技术指标验证模型的判断。比如模型看多同时5日均线上穿20日均线信号置信度就可以提高如果模型看多但RSI进入超买区报告里应该追加一条警示。7.4 合规与风险提示不建议越过哪些边界多AI看一张图这个功能产品价值确实存在但有几条边界不建议越过。第一不要在界面里出现“AI建议直接买入”“AI预测必涨”这类表述。模型的输出本质是概率性推测不是事实。所有结论都必须标注“仅供研究参考不构成投资建议”。第二不要把模型的置信度渲染成收益预测。confidence0.9只能说明模型对自己判断的确定性高不能翻译成“90%概率赚钱”。第三API Key不要提交到git仓库。建议在.gitignore中加入.env并使用环境变量或密钥管理服务。第四如果产品面向公众用户需要在显著位置提示用户AI视觉分析会受到图片清晰度、指标设置、数据范围等因素影响结论可能不够准确。第五回测不能说明未来。用历史K线图测出“模型判断很准”之后不要直接上实盘。视觉模型容易记住训练数据里的形态但历史规律会随市场结构变化失效。8. 从演示到可用最该盯住哪几件事整套代码写完能跑通多模型分析之后最值得花时间的不是继续增加模型数量而是做三件事。第一件事建立自己的评估集。准备20到30张不同类型的K线图包含明显的上升趋势、下降趋势、横盘震荡、长上影线、放量突破等场景记录每个模型的输出。不要只看最终结论还要看关键点位、风险点、分析理由是不是符合常识。没有评估集你永远不知道模型什么时候在胡说。第二件事锁死输出协议。多模型接入最大的风险不是模型能力不够而是输出格式不稳定。一个模型返回了完整JSON另一个模型在JSON前后加了大量解释汇总层就被迫做各种容错。应该让所有模型共用同一份严格提示词模板并用测试用例覆盖“解析失败”“字段缺失”等异常分支。第三件事先在辅助场景运行。把多AI分析结果放到日报、周报、研究草稿这类“有人工复核”的功能里观察一段时间再考虑是否把结论接入自动提醒或策略信号。直接接入交易执行系统现阶段风险远大于收益。这套多模型看图的实现本质上是一个很典型的“模型编排 结构化输出 聚合决策”工程问题。即使以后换成更新的模型只要保持统一接口、统一输出协议、统一汇总逻辑迁移成本都不会太大。对想入门的开发者来说这也是一个把大模型能力从“聊天玩具”往“可计算工具”推进的合适练习项目。