企业官网建设流程全解析

AI量化交易最近讨论最多的方向，已经从“用AI写策略”变成了“让AI直接读图并给出交易分析”。所谓多个AI同时看一张图，是指把同一张K线图、分时图或技术指标图，同时发送给多个大语言模型，让每个模型独立输出趋势判断、关键点位和风险提示，再由系统汇总成一份综合结论。这种做法的价值不在于某一个模型有多强，而在于多个模型在同一张图上互相印证、彼此纠偏。标题中提到的孙越AI，正是这类多模型量化分析工具的一个产品化形态。下面用一套最小可运行实现，拆解它的核心链路：图表预处理、多模型并行调用、结果解析和综合汇总，并给出本地验证和生产落地时最容易踩的坑。

1. 多AI看一张图，先想清楚它解决什么问题

1.1 单模型分析量化图表的局限性

传统量化分析有两种常见做法。一种靠人工盯盘，看K线形态、均线位置、成交量变化，然后凭经验判断。另一种靠规则引擎，把“MACD金叉”“突破20日均线”这类条件写成程序，满足条件就触发信号。

前者的问题是主观、容易疲劳，同一个人在不同时间看同一张图，结论都可能不一样。后者的问题更隐蔽：规则引擎只能识别写死的形态，无法理解形态背后的市场氛围，也没有办法解释“为什么这里可能是支撑位”。

大模型出现后，很多人尝试让模型直接看K线截图，用自然语言输出分析。效果比预想好，但单独用一个模型做这件事，会暴露出四类问题：

1. 视觉理解偏差。模型对K线实体、影线、均线交叉、成交量放量的理解不一定准确，经常会把长上影线看成强势突破。
2. 上下文遗漏。K线图信息密度高，模型在长上下文里可能漏掉关键位置，比如前期高点、密集成交区。
3. 过度自信。即使模型判断错了，它也会用很确定的语气输出“大概率上涨”，缺少自我怀疑。
4. 风格偏置。同一个任务，有的模型习惯看多，有的模型习惯看空，单模型会导致结论体系性偏移。

所以“多个AI同时看一张图”不是炫技，而是用模型组合来对冲单模型的不确定性。

1.2 多模型并行分析的产品形态

把多个AI同时看一张图拆开看，本质是“多智能体评审”。系统把同一张图同时分发给多个大模型，每个模型独立输出结构化报告，再由汇总模块生成综合结论。

产品端呈现通常是这样：

• 用户上传一张K线截图或分时图。
• 系统显示多个模型的分析卡片，每个卡片包含趋势判断、置信度、关键点位、风险点。
• 顶部显示汇总结论，标出哪些模型观点一致，哪些模型存在分歧。

这里的核心并不在于“谁对谁错”，而在于两点：共识部分是否稳定，分歧部分能否提示风险。如果三个模型都看多，结论可信度就比单个模型看多高得多。如果两个看多、一个看空，系统应该把这个分歧明确提示出来，而不是强行取平均。

这里的预期收益不是“准确率提升几个点”，而是风险表达更完整：单模型漏掉的风险，另一个模型可能会补上。

1.3 这个功能适合用在什么场景

并不是所有量化场景都适合让AI看图表。适合的是那些“需要快速生成分析草稿、需要多视角解读、需要人工复核”的场景。

最后一行尤其重要。现阶段让多个AI看一张图，适合作为“分析助手”，不适合直接接管交易。原因很简单：模型没有实时行情权限，视觉推理本身存在误差，而且它无法感知盘口深度、资金流向、突发消息等非图表信息。

2. 系统设计与技术拆分

2.1 整体处理链路：图表输入、模型分发、结果汇总

多AI看一张图，技术链路可以拆成五步：

1. 接收用户上传的图片。
2. 对图片做统一预处理：裁剪多余区域、缩放分辨率、转base64编码。
3. 把同一张图片和同一份提示词，并发发送给多个模型。
4. 每个模型返回自然语言或JSON文本，解析成统一结构。
5. 汇总模块对多个结构化结果做投票、合并和风险提示，最后生成报告。

这条链路最容易被忽略的环节是第一步和第二步。很多实现把用户上传的原图直接发给模型，图片分辨率过高会导致token消耗暴涨，过低又会导致模型看不清K线形态。后面会单独讲预处理参数。

分发环节要注意“并发”和“超时”。多模型串行调用，总耗时等于所有模型耗时相加，体验会很差。正确做法是并发调用，同时给每个模型设置独立的超时时间，避免某个模型卡住拖垮整个任务。

汇总环节不是简单拼接。如果只是把多个模型的文字结果堆在一起，用户还得自己对比，体验和看多个网页没有区别。汇总必须解决“观点是否一致”“分歧在哪里”“最终建议是什么”这三个问题。

2.2 模型供应商层：用兼容接口屏蔽差异

不同模型厂商的协议不统一，如果为每个模型写一套单独调用代码，维护成本会很高。

实践中建议统一使用OpenAI兼容协议。当前主流模型的API服务，多数都提供OpenAI兼容的endpoint，只需要把base_url改成对应厂商的地址，同时替换api_key和model名称即可。

在设计模型供应商层时，要抽象出一个基础接口：

这样做的好处是后续接入新模型时，不需要改动上层调度和汇总逻辑，只需要新增一个配置项或一个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.txt：

openai>=1.35.0 Pillow>=10.0.0 requests>=2.31.0 PyYAML>=6.0.1 matplotlib>=3.8.0

逐项说明用途：

安装后确认版本，特别要注意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

各文件职责：

这种结构在本地开发时够用，上线时可以把providers和core拆成独立服务。

3.3 配置文件

在项目根目录创建config.yaml：

models: - 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

这里的关键参数说明：

注意：不同厂商的base_url会变化，落地前应以对应服务商当前文档为准。上述地址是常见的兼容接口格式，只用于示例。

4. 核心代码实现

4.1 图表预处理：裁剪、缩放、转Base64

图片预处理是很容易被忽略的一步。用户上传的截图可能是整个屏幕，包含广告栏、标题、水印，这些噪声会干扰模型识别，也会浪费token。

建议处理顺序是：

1. 打开图片。
2. 可选裁剪，去掉上下边缘非图表区域。
3. 按最长边缩放到max_image_size。
4. 转成RGB模式，避免RGBA图片带透明通道导致编码异常。
5. 压缩为JPEG格式，质量设为90。
6. 转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, format="JPEG", quality=quality) 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_key=api_key, base_url=base_url, timeout=timeout) self.model = model self.temperature = temperature def analyze_chart(self, image_base64: str, prompt: str) -> str: response = self.client.chat.completions.create( model=self.model, temperature=self.temperature, messages=[ { "role": "user", "content": [ {"type": "text", "text": prompt}, { "type": "image_url", "image_url": { "url": f"data: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_workers=4): results = {} with ThreadPoolExecutor(max_workers=max_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, timeout=90): 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, timeout=90)的timeout是整个等待的总超时，不是每个任务的超时。所以配置里的timeout要同时考虑模型数量和模型本身耗时。

单个模型失败时，不要直接抛异常让整个程序退出。更合理的做法是把错误记录下来，让其他模型的结果继续参与汇总。失败模型的卡片显示“当前不可用”，用户和管理员都能看到。

4.4 结构化结果解析与汇总

解析完成之后，进入汇总阶段。汇总模块要解决四个问题：

1. 趋势判断不一致时怎么办。
2. 置信度如何合并。
3. 关键点位重复和冲突如何处理。
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, tolerance=0.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, key=lambda 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里的tolerance=0.01表示价格相差小于0.01的点位视为同一个位置。实际项目要根据交易品种调整这个值：股票价格可能在几十元到几百元，0.01太严格；数字货币价格可能在几万美元，0.01又太严格。更合理的做法是按价格比例去重，比如abs(price - p) / p < 0.001。

5. 运行验证与结果分析

5.1 用同一张图跑一次多模型分析

为了本地验证，先写一个生成样例K线图的脚本。sample/kline_sample.py：

import 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(start="2025-01-01", periods=n, freq="D") 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, }, index=dates) mpf.plot(df, type="candle", volume=True, style="charles", savefig="sample/kline_sample.png")

运行：

python sample/kline_sample.py

生成sample/kline_sample.png后，写一个完整的入口main.py：

import 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( name=item["name"], api_key_env=item["api_key_env"], base_url=item["base_url"], model=item["model"], timeout=item.get("timeout", 60), ) providers.append(provider) return providers def main(image_path, config_path="config.yaml"): with open(config_path, "r", encoding="utf-8") as f: config = yaml.safe_load(f) providers = load_providers(config) image_base64 = preprocess_image( image_path, max_size=config["models"][0].get("max_image_size", 1024), ) prompt = build_prompt() raw_results = analyze_with_all_models( providers, image_base64, prompt, max_workers=config.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_ascii=False, indent=2)) 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只是“少数服从多数”，没有实际决策价值。

第二条规则：置信度只能相对比较。模型说confidence=0.9，不代表真实概率是90%。但同一个模型对两张图分别给出0.9和0.5时，0.9的那张图通常更值得关注。

第三条规则：空跑测试不可少。拿一张没有明显趋势的横盘图去测，如果多个模型仍然编造出“强支撑位”“突破压力位”，说明提示词太容易被带偏，或者模型本身幻觉严重。应该让模型在不确定性高时输出uncertain，而不是硬给一个多空判断。

5.3 一致性指标

为了衡量“多个AI是否真的在看同一张图”，可以引入简单的一致性指标。

这些指标不需要写得很复杂。如果项目刚起步，只需要输出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,前缀。
• 图片分辨率过大，超出服务端限额。

检查方式：

1. 打印实际请求中的base_url、model、api_key_env。
2. 确认图片编码后字符串以/9j/开头，这是JPEG的base64特征前缀。
3. 把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 排错顺序清单

遇到问题时，按以下顺序排查，能省掉大量无意义的日志翻找：

这个清单适用于大多数“多模型看一张图”场景。核心思路是从上游到下游逐段验证：输入有问题就先解决输入，不要先去猜模型是不是疯了。

7. 生产化落地要点

7.1 学习环境与生产环境的差异

本地能跑通和线上能稳定运行是两回事。区别集中在资源、容错和可观测性上。

生产环境还要考虑一件事：图片上传后不能无限制保留。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预测必涨”这类表述。模型的输出本质是概率性推测，不是事实。所有结论都必须标注“仅供研究参考，不构成投资建议”。

第二，不要把模型的置信度渲染成收益预测。confidence=0.9只能说明模型对自己判断的确定性高，不能翻译成“90%概率赚钱”。

第三，API Key不要提交到git仓库。建议在.gitignore中加入.env，并使用环境变量或密钥管理服务。

第四，如果产品面向公众用户，需要在显著位置提示用户：AI视觉分析会受到图片清晰度、指标设置、数据范围等因素影响，结论可能不够准确。

第五，回测不能说明未来。用历史K线图测出“模型判断很准”之后，不要直接上实盘。视觉模型容易记住训练数据里的形态，但历史规律会随市场结构变化失效。

8. 从演示到可用，最该盯住哪几件事

整套代码写完，能跑通多模型分析之后，最值得花时间的不是继续增加模型数量，而是做三件事。

第一件事，建立自己的评估集。准备20到30张不同类型的K线图，包含明显的上升趋势、下降趋势、横盘震荡、长上影线、放量突破等场景，记录每个模型的输出。不要只看最终结论，还要看关键点位、风险点、分析理由是不是符合常识。没有评估集，你永远不知道模型什么时候在胡说。

第二件事，锁死输出协议。多模型接入最大的风险不是模型能力不够，而是输出格式不稳定。一个模型返回了完整JSON，另一个模型在JSON前后加了大量解释，汇总层就被迫做各种容错。应该让所有模型共用同一份严格提示词模板，并用测试用例覆盖“解析失败”“字段缺失”等异常分支。

第三件事，先在辅助场景运行。把多AI分析结果放到日报、周报、研究草稿这类“有人工复核”的功能里，观察一段时间，再考虑是否把结论接入自动提醒或策略信号。直接接入交易执行系统，现阶段风险远大于收益。

这套多模型看图的实现，本质上是一个很典型的“模型编排 + 结构化输出 + 聚合决策”工程问题。即使以后换成更新的模型，只要保持统一接口、统一输出协议、统一汇总逻辑，迁移成本都不会太大。对想入门的开发者来说，这也是一个把大模型能力从“聊天玩具”往“可计算工具”推进的合适练习项目。