最近在做 AI 量化交易工具时，遇到了一个很有意思的需求：单一大模型看图分析 K 线，结果往往不够稳定。有的模型擅长识别形态，有的模型更关注均线关系，还有的模型对成交量变化更敏感。于是我们做了一项新功能：让多个 AI 同时看同一张图，各自独立给出分析，最终汇总为一份综合研判结果。今天把这项功能的实现思路、关键代码和踩坑记录完整整理出来，希望对你构建自己的 AI 量化分析系统有帮助。
本文适合有 Python 基础的开发者，涉及的内容包括：多模态大模型 API 调用、K 线图表生成、多模型并行分析的任务拆分、结果聚合策略，以及在实际工程中容易忽略的稳定性和成本问题。
1. 背景与核心概念
1.1 单模型看图分析的不足
传统量化策略分析师依赖 K 线、均线、MACD、成交量等指标做判断。大模型出现后，很多人尝试直接把 K 线截图丢给 GPT-4V、Claude、Gemini 这类多模态模型，让它们“看图说话”。
单模型分析有一个明显问题：不同模型的能力侧重点差异很大。有的模型对图表中的文字、数字比较敏感，能准确读出 K 线数据；有的模型长于趋势判断，但对具体数值容易出错；还有的模型在形态识别上表现不错，却忽略成交量配合。
这意味着，单一大模型的分析结果往往带有“盲区”。如果多个 AI 同时看同一张图，从不同角度给出各自分析，再把这些分析做交叉验证和综合汇总，就能显著提升分析的稳定性和参考价值。
1.2 多 AI 并行分析的基本思路
所谓“多个 AI 同时看一张图”，本质上是：
- 统一生成一张包含核心行情信息的 K 线图。
- 把这张图同时发给多个多模态 AI 模型。
- 每个模型独立输出对当前行情的分析，互不干扰。
- 系统收集所有模型的分析结果，按结构要求汇总输出。
示意图可以理解为下面的简单流程：
K线图文件 -> 模型A(趋势判断) -> 分析A -> 模型B(形态识别) -> 分析B -> 汇总引擎 -> 综合研判报告 -> 模型C(量价关系) -> 分析C这里的核心不是“多个模型聊天”，而是彼此隔离的独立分析通道，最后再统一汇总。这样设计的好处是避免模型之间的观点互相污染，也能保证分析结果可回溯——每一份分析都能找到对应的模型来源。
1.3 典型应用场景
多 AI 联合看盘功能可以应用在几个实际场景中：
- 盘前复盘：每天开盘前，让多个 AI 对昨日 K 线图做独立复盘，生成多视角解读。
- 个股异动分析：发现某只股票短期涨幅异常时，让多个 AI 分别分析原因和后续风险。
- 策略信号验证：量化策略产生买点信号后，用多 AI 分析来辅助确认信号质量。
- 学习辅助：新手可以把多 AI 的分析结果作为参考，理解不同流派对同一张图的不同解释。
2. 环境准备与版本说明
本文示例代码以 Python 3.10+ 为基础，核心依赖如下：
pip install akshare pandas matplotlib openai anthropic需要说明的是，模型 API 的版本迭代速度很快，下面的代码以常见的 OpenAI、Anthropic 接口为例演示整体思路。你在实际项目中要根据自己使用的模型官方文档调整请求参数，不要照抄后直接用于生产环境。
操作系统以 Windows 10/11 或 Ubuntu 20.04+ 均可，代码不依赖某个特定系统特性。
这里用到的模型配置如下：
| 模型 | 用途 | 说明 |
|---|---|---|
| 模型A | 趋势与均线判断 | 擅长整体趋势解读 |
| 模型B | K线形态识别 | 擅长识别锤子线、吞没形态等 |
| 模型C | 量价关系分析 | 擅长结合成交量判断 |
实际项目中，你可以替换为任意支持图像输入的多模态模型。建议至少准备两个不同厂商的模型，避免因为单家模型服务异常导致整个功能不可用。
3. 核心原理拆解
3.1 为什么要把分析角色拆开
多 AI 并行分析并不是简单地“让多个模型重复做同一件事”，而是为每个模型分配一个明确的分析角色。
比如：
- 模型A的提示词强调“只关注趋势和均线系统，忽略其他细节”。
- 模型B的提示词强调“只识别经典K线形态，给出出现位置”。
- 模型C的提示词强调“关注成交量与价格的关系”。
这样做的好处有三点：
- 减少幻觉。模型在明确的角色约束下，不会产生大而全但模糊的回答。
- 提升可读性。汇总结果时，每个维度清晰独立。
- 方便交叉验证。当多个模型对同一现象给出相似结论时，可信度更高。
3.2 图像输入的处理方式
大模型读取图片有两种常见方式：
- Base64 编码：把图片文件编码为 Base64 字符串，放入请求消息中。
- 图片 URL：把图片上传到临时图床，把 URL 传给模型。
本地量化分析场景推荐使用 Base64 方式，因为不依赖外网图床，更安全。但要注意，Base64 会把图片体积膨胀约 33%，请求体可能较大，需要合理压缩图片大小。
3.3 提示词设计框架
多 AI 看图分析的提示词，建议统一采用以下框架：
你是[角色]。请分析这张K线图。 重点关注：[具体分析维度1]、[具体分析维度2]。 回答要求： 1. 用中文回答，控制在200字以内。 2. 先给出结论，再简要说明依据。 3. 如果图片不清晰或无法判断，明确说明。这里的核心是限制回答范围。不要问模型“这张图说明什么”，而要问“从趋势角度看，这张图呈现了什么状态”。模型在约束条件下输出的内容更可控。
4. 完整实战案例：构建多 AI 联合看图分析系统
下面我们逐步实现一个最小可用的多 AI 看图分析系统。整体分为四个部分：
- 使用 akshare 获取行情数据。
- 使用 matplotlib 生成带均线和成交量的 K 线图。
- 调用多个多模态大模型 API 分别分析图片。
- 汇总各模型输出，生成综合报告。
4.1 创建项目结构
首先创建项目目录：
multi_ai_analysis/ ├── chart_utils.py # 生成K线图 ├── ai_clients.py # 封装多模型调用 ├── aggregator.py # 汇总分析结果 ├── main.py # 主入口 └── config.py # 配置文件4.2 生成 K 线图
我们使用 akshare 获取 A 股或指数的日线数据，然后用 matplotlib 绘制带均线和成交量的 K 线图。
# 文件路径：multi_ai_analysis/chart_utils.py # -*- coding: utf-8 -*- """ 生成K线图：包含K线、5日/20日均线、成交量 """ import akshare as ak import pandas as pd import matplotlib import matplotlib.pyplot as plt from matplotlib.patches import Rectangle matplotlib.use("Agg") # 使用无界面后端，适合服务器环境 plt.rcParams["font.sans-serif"] = ["SimHei", "Microsoft YaHei", "PingFang SC"] plt.rcParams["axes.unicode_minus"] = False def fetch_daily_data(symbol: str, days: int = 60) -> pd.DataFrame: """ 获取日线数据，返回包含 open/high/low/close/volume 的DataFrame。 symbol 示例：sh600519，表示上证贵州茅台 """ # 这里以股票日线为例，实际可以根据需要获取指数或期货数据 df = ak.stock_zh_a_hist(symbol=symbol, period="daily", adjust="qfq") df = df.tail(days).reset_index(drop=True) df.columns = ["date", "open", "close", "high", "low", "volume", "amount", "amplitude", "pct_change", "change", "turnover"] return df def plot_kline(df: pd.DataFrame, save_path: str) -> str: """ 绘制K线图，并保存到 save_path。 """ df["ma5"] = df["close"].rolling(5).mean() df["ma20"] = df["close"].rolling(20).mean() fig, ax = plt.subplots(2, 1, figsize=(14, 10), gridspec_kw={"height_ratios": [3, 1]}) plt.subplots_adjust(hspace=0.05) # 绘制K线 for idx, row in df.iterrows(): color = "red" if row["close"] >= row["open"] else "green" # 影线 ax[0].plot([idx, idx], [row["low"], row["high"]], color=color, linewidth=0.8) # 实体 body_bottom = min(row["open"], row["close"]) body_height = abs(row["close"] - row["open"]) if body_height < 1e-10: body_height = 0.01 rect = Rectangle((idx - 0.3, body_bottom), 0.6, body_height, facecolor=color, edgecolor=color) ax[0].add_patch(rect) # 均线 ax[0].plot(df["ma5"], label="MA5", color="orange", linewidth=1.2) ax[0].plot(df["ma20"], label="MA20", color="blue", linewidth=1.2) ax[0].legend() ax[0].set_title("KLine Analysis") ax[0].set_ylabel("Price") ax[0].grid(True, alpha=0.3) # 成交量 vol_bars = ax[1].bar(df.index, df["volume"], color=["red" if c >= o else "green" for o, c in zip(df["open"], df["close"])], alpha=0.6) ax[1].set_ylabel("Volume") ax[1].grid(True, alpha=0.3) # 标注日期刻度，避免太密集 tick_step = max(1, len(df) // 8) tick_indices = list(range(0, len(df), tick_step)) ax[0].set_xticks(tick_indices) ax[0].set_xticklabels([df.iloc[i]["date"] for i in tick_indices], rotation=30) ax[1].set_xticks(tick_indices) ax[1].set_xticklabels([df.iloc[i]["date"] for i in tick_indices], rotation=30) plt.savefig(save_path, dpi=120, bbox_inches="tight") plt.close(fig) return save_path这里有几个细节需要解释：
- 成交量使用红绿柱表示涨跌，与 K 线保持一致。
- 均线只保留 MA5 和 MA20，避免图面过于复杂，方便大模型识别核心信息。
- 日期刻度做了稀疏处理，防止重叠影响模型读取。
4.3 封装多模型调用
接下来封装 AI 客户端的调用逻辑。不同的模型使用的 SDK 不同，我们用一个统一的接口来封装。
# 文件路径：multi_ai_analysis/ai_clients.py # -*- coding: utf-8 -*- """ 封装多个多模态大模型的调用接口 """ import base64 import os from openai import OpenAI import anthropic def _encode_image(image_path: str) -> str: """将图片文件编码为 Base64 字符串""" with open(image_path, "rb") as f: return base64.b64encode(f.read()).decode("utf-8") class BaseAIClient: """统一的AI客户端基类""" def analyze(self, image_path: str, prompt: str) -> str: raise NotImplementedError class OpenAIClient(BaseAIClient): """OpenAI GPT-4o / GPT-4V 客户端示例""" def __init__(self, api_key: str = None, model: str = "gpt-4o"): self.client = OpenAI( api_key=api_key or os.getenv("OPENAI_API_KEY") ) self.model = model def analyze(self, image_path: str, prompt: str) -> str: b64_image = _encode_image(image_path) response = self.client.chat.completions.create( model=self.model, messages=[ { "role": "user", "content": [ {"type": "text", "text": prompt}, { "type": "image_url", "image_url": { "url": f"data:image/png;base64,{b64_image}" } } ] } ], max_tokens=500, temperature=0.3 ) return response.choices[0].message.content class AnthropicClient(BaseAIClient): """Anthropic Claude 客户端示例""" def __init__(self, api_key: str = None, model: str = "claude-3-5-sonnet-20241022"): self.client = anthropic.Anthropic( api_key=api_key or os.getenv("ANTHROPIC_API_KEY") ) self.model = model def analyze(self, image_path: str, prompt: str) -> str: b64_image = _encode_image(image_path) response = self.client.messages.create( model=self.model, max_tokens=500, temperature=0.3, messages=[ { "role": "user", "content": [ { "type": "image", "source": { "type": "base64", "media_type": "image/png", "data": b64_image } }, {"type": "text", "text": prompt} ] } ] ) return "".join(block.text for block in response.content)如果你使用其他模型，比如 Google Gemini、智谱 GLM-4V、通义千问 VL 等，思路完全一致：把图片转为 Base64，按照对应模型的请求格式发送即可。
4.4 定义各 AI 的分析角色与提示词
同一个模型也可以承担不同角色，但为了让多 AI 效果更好，建议为每个模型分配固定角色。这里我们定义三组提示词：
# 文件路径：multi_ai_analysis/config.py # -*- coding: utf-8 -*- """ 模型角色与提示词配置 """ TREND_PROMPT = """ 你是一位趋势分析专家。请分析这张K线图： 1. 当前处于上升趋势、下降趋势还是震荡趋势？ 2. MA5 与 MA20 的位置关系是什么？ 3. 最近5根K线的整体走势如何？ 4. 给出未来3-5个交易日的趋势判断（偏向谨慎表述）。 要求：200字以内，先给结论，再给依据。 """ PATTERN_PROMPT = """ 你是一位K线形态识别专家。请分析这张K线图： 1. 图中是否存在经典K线形态（如锤子线、吞没形态、十字星等）？ 2. 如果存在，出现在图中什么位置？ 3. 这些形态在当前位置意味着什么？ 4. 给出关键支撑位和阻力位的粗略判断。 要求：200字以内，先给结论，再给依据。 """ VOLUME_PRICE_PROMPT = """ 你是一位量价分析专家。请分析这张K线图： 1. 近期成交量与价格走势是否配合？ 2. 是否出现放量上涨、缩量回调或放量下跌等信号？ 3. 从量价关系角度看，当前多空力量对比如何？ 4. 给出量价层面的风险提示。 要求：200字以内，先给结论，再给依据。 """ # 为每个模型指定角色 MODEL_ROLES = [ { "name": "趋势分析", "prompt": TREND_PROMPT, }, { "name": "形态识别", "prompt": PATTERN_PROMPT, }, { "name": "量价分析", "prompt": VOLUME_PRICE_PROMPT, }, ]4.5 汇总多模型结果
汇总逻辑是整个系统的关键。我们不希望把三个模型的回答简单拼接，而是要做结构化整理。
# 文件路径：multi_ai_analysis/aggregator.py # -*- coding: utf-8 -*- """ 汇总多个AI的分析结果，生成综合报告 """ import json def aggregate_results(results: list) -> dict: """ results 格式： [ {"role_name": "趋势分析", "model_name": "gpt-4o", "content": "..."}, {"role_name": "形态识别", "model_name": "claude-...", "content": "..."}, {"role_name": "量价分析", "model_name": "gpt-4o", "content": "..."}, ] 返回汇总后的结构化结果 """ summary = { "total_models": len(results), "success_models": 0, "details": [], "combined_text": "", } for item in results: if item["content"] and not item["content"].startswith("[错误]"): summary["success_models"] += 1 detail = { "role": item["role_name"], "model": item["model_name"], "analysis": item["content"], } summary["details"].append(detail) # 拼接纯文本，方便后续人工阅读或二次处理 parts = [] for detail in summary["details"]: parts.append(f"【{detail['role']}】\n{detail['analysis']}") summary["combined_text"] = "\n\n".join(parts) # 简单的一致性判断：如果成功模型少于2个，降级提示 if summary["success_models"] < 2: summary["consensus"] = "样本不足，无法形成有效共识" else: summary["consensus"] = "多个模型已完成独立分析，请结合各维度结论综合判断" return summary def save_report(summary: dict, output_path: str = "report.json") -> None: """保存汇总报告到本地文件""" with open(output_path, "w", encoding="utf-8") as f: json.dump(summary, f, ensure_ascii=False, indent=2)这里做了一个简单的降级处理：如果多个模型中只有 1 个返回成功，说明系统不稳定，不强行生成共识结论。这个设计在实际工程中很重要，因为大模型 API 偶尔会超时或返回空内容。
4.6 主流程
在主流程中串联所有步骤：
# 文件路径：multi_ai_analysis/main.py # -*- coding: utf-8 -*- """ 主入口：多AI联合看图分析 """ import os from chart_utils import fetch_daily_data, plot_kline from ai_clients import OpenAIClient, AnthropicClient from config import MODEL_ROLES from aggregator import aggregate_results, save_report def build_clients(): """ 根据环境变量初始化模型客户端。 这里演示同时使用 OpenAI 和 Anthropic 的模型。 """ clients = {} if os.getenv("OPENAI_API_KEY"): clients["gpt-4o"] = OpenAIClient() if os.getenv("ANTHROPIC_API_KEY"): clients["claude-3-5-sonnet"] = AnthropicClient() return clients def run_analysis(symbol: str, days: int = 60): # 1. 准备数据与图表 print(f"[1/4] 获取 {symbol} 行情数据...") df = fetch_daily_data(symbol, days) chart_path = "kline.png" plot_kline(df, chart_path) print(f"[1/4] K线图已生成: {chart_path}") # 2. 初始化模型客户端 print("[2/4] 初始化AI模型客户端...") clients = build_clients() if not clients: raise RuntimeError("未配置任何API Key，请设置 OPENAI_API_KEY 或 ANTHROPIC_API_KEY") # 3. 让每个模型处理每个角色 # 实际项目中可以把每个角色的请求放到线程池中并行执行 print("[3/4] 多个AI并行分析中...") results = [] for role in MODEL_ROLES: # 这里简单轮询使用客户端，生产环境可增加负载均衡 client_name = list(clients.keys())[len(results) % len(clients)] client = clients[client_name] try: content = client.analyze(chart_path, role["prompt"]) except Exception as e: content = f"[错误] {str(e)}" results.append({ "role_name": role["name"], "model_name": client_name, "content": content, }) print(f" - {role['name']} 完成") # 4. 汇总输出 print("[4/4] 汇总分析结果...") summary = aggregate_results(results) save_report(summary, "analysis_report.json") print(f"分析报告已保存: analysis_report.json") print("=" * 40) print(summary["combined_text"]) if __name__ == "__main__": # 示例：分析上证指数 run_analysis(symbol="000001", days=60)4.7 运行与验证
在项目根目录执行：
export OPENAI_API_KEY="你的OpenAI Key" export ANTHROPIC_API_KEY="你的Anthropic Key" python main.py预期输出如下（内容因模型不同会有差异）：
[1/4] 获取 000001 行情数据... [1/4] K线图已生成: kline.png [2/4] 初始化AI模型客户端... [3/4] 多个AI并行分析中... - 趋势分析 完成 - 形态识别 完成 - 量价分析 完成 [4/4] 汇总分析结果... 分析报告已保存: analysis_report.json ======================================== 【趋势分析】 当前处于震荡偏多趋势，MA5 上穿 MA20 ... 【形态识别】 近期出现十字星形态，结合位置看，存在变盘可能 ... 【量价分析】 成交量整体持平，未见明显放量 ...同时，当前目录下会产生：
kline.png：输入给模型的 K 线图。analysis_report.json：结构化分析报告。
4.8 并行优化
上面的主流程是串行调用三个模型，实际使用中速度较慢。推荐改为线程池并发请求：
from concurrent.futures import ThreadPoolExecutor, as_completed def run_parallel_analysis(clients, chart_path): results = [] total = len(MODEL_ROLES) def _request(role): client_name = list(clients.keys())[len(results) % len(clients)] client = clients[client_name] try: content = client.analyze(chart_path, role["prompt"]) except Exception as e: content = f"[错误] {str(e)}" return { "role_name": role["name"], "model_name": client_name, "content": content, } with ThreadPoolExecutor(max_workers=total) as executor: future_map = {executor.submit(_request, role): role for role in MODEL_ROLES} for future in as_completed(future_map): results.append(future.result()) return results并行后，整体耗时取决于最慢的那个模型调用，而不是三个模型耗时的总和，能明显提升体验。
5. 常见问题与排查思路
5.1 模型返回内容为空
| 问题现象 | 常见原因 | 解决思路 |
|---|---|---|
| 返回 content 为 None 或空字符串 | 图片解析失败、请求被安全策略拦截 | 检查图片是否为标准PNG，尺寸是否过大；打印完整响应结构，确认返回字段名 |
| 返回内容包含“抱歉，无法分析” | 提示词不够具体，模型未识别为有效任务 | 强化角色设定，明确要求分析维度 |
5.2 图片生成后模型识别困难
如果模型看不清 K 线图上的信息，通常不是模型问题，而是图表本身信息密度过高的原因。
我们建议：
- 把图片宽度控制在 1000 像素左右，不要用超大原图。
- 均线数量控制在 2~3 条，不要叠加太多指标。
- 中文标注如果识别不好，改成英文标注或阿拉伯数字。
- 压缩图片质量到 80% 左右，减少 Base64 体积。
5.3 不同模型对同一张图结论不一致
这是正常现象，也是多 AI 分析的价值所在。如果结论完全一致，反而说明模型可能受到了共同训练数据的影响。
处理策略：
- 在汇总报告中明确标注每个模型的“角色”，不要求所有模型回答相同问题。
- 对结论做交叉验证：如果趋势分析说“上升”，量价分析说“缩量上行需谨慎”，综合结论应体现这种分歧。
- 不建议用投票方式简单决定最终结论，量化交易中需要的是多维度参考，而不是民主表决。
5.4 API 调用超时
多模态请求比纯文本请求慢很多，尤其是图片较大的时候：
| 问题现象 | 常见原因 | 解决思路 |
|---|---|---|
| 请求超过 30 秒无响应 | 图片体积过大、网络不稳定 | 压缩图片、设置更高超时时间 |
| 部分模型失败影响整体 | 没有做超时隔离 | 使用 as_completed + Future.timeout 控制单项等待时间 |
建议在客户端中增加超时参数，例如 OpenAI 的timeout=60，Claude 的timeout=120，具体数值根据网络情况调整。
5.5 成本失控
多 AI 分析意味着每次完整分析要调用多次模型，成本是单次分析的数倍。控制成本的方法包括：
- 控制每次请求的
max_tokens，不需要模型写长篇分析。 - 对图片做降采样，缩小图片体积。
- 设置每日调用上限，按用户粒度限流。
- 使用模型缓存层，对相同 K 线图不重复调用。
6. 最佳实践与工程建议
6.1 提示词版本管理
AI 量化分析系统的提示词会频繁调整。建议把提示词放入独立的配置文件或数据库中，而不是硬编码在代码里。每次修改提示词后记录版本号，方便回滚对比。
6.2 分析结果的可追溯性
每份分析报告必须包含：
- 生成时间。
- 使用的模型名称与版本。
- 提示词版本。
- K 线图数据的时间范围。
这样当分析结论出现严重偏差时，你可以回到当时的输入条件，定位是哪一层出了问题。
6.3 熔断与降级
生产环境中，模型 API 不可用是常态。一定要设计降级策略：
- 如果主要模型失败，自动切换备用模型。
- 如果所有模型都失败，返回“AI分析服务暂不可用”，不得返回空报告。
- 监控模型调用成功率，连续失败超过阈值时触发告警。
6.4 安全边界
AI 分析结果只能作为参考，不能替代投资决策。在 UI 或报告中应明确标注“AI 生成内容仅供参考，不构成投资建议”。这一点既是合规要求，也是对用户负责。
6.5 缓存策略
对于同一标的、同一时间周期的 K 线图，在短时间内分析结果差异不大。可以按“标的+周期+日期”做缓存，当天重复请求直接返回缓存结果，减少 API 成本和延迟。
6.6 日志与监控
每个模型的请求都应该记录：
logger.info({ "event": "ai_analysis_request", "symbol": symbol, "model": client_name, "role": role_name, "cost_tokens": tokens_used, "latency_ms": latency_ms, "success": success, })这些数据后续可以用于分析模型效果、统计成本、排查故障。
7. 总结与下一步学习方向
本文实现了一个“多个 AI 同时看一张图并给出各自分析”的完整功能：先生成 K 线图，再通过多线程方式将图片分别发送给不同角色的大模型，最后汇总分析并输出结构化工单。项目核心代码可以在本地直接运行，替换 API Key 后即可作为自己的 AI 量化分析工具使用。
接下来你可以从这几个方向继续深入：
- 接入更多金融数据源，例如期货、加密货币、港美股数据。
- 将分析结果接入自己的量化策略回测系统，做信号叠加。
- 增加更丰富的图表元素，如布林带、MACD，但要注意控制信息密度。
- 使用向量数据库存储历史分析报告，构建 AI 量化投研知识库。
多 AI 分析不是让模型替你决策，而是让不同视角的信息帮你减少盲区。真正落地时，模型的稳定输出、成本控制和结果回溯往往比“模型本身有多聪明”更重要。
希望这篇实战教程能给你带来启发。如果你在实现过程中遇到模型 API 兼容、图表识别或结果汇总方面的问题，欢迎留言交流。
