Files
auv/backend/services/ai_service.py
T
Sakurasan e24b1a360b feat: 优化每日分析提示词模板并修复分段生成链路
提示词:
- SYSTEM_PROMPT 重写为可执行原则(事实优先/时间锚定/判断挂数字/禁用套话)+输出格式契约
- DAILY_ANALYSIS_PROMPT 重排注意力:取数指令置顶(要求同轮并行调用 4 个核心工具)、
  写作规约 8 条、环比快照居中、前日报告全文置底并改为「今日数据优先+核对昨日展望」
- 新增单位口径、符号规范(+/- 是前端红涨绿跌着色依据)、不荐股等硬规约
- 移除已废弃的「今日定调」摘要规则(与分段指令、前端渲染互相冲突)
- 八章重构:板块资金流独立为「四、资金与筹码」,重复的「关注方向+下个交易日建议」
  合并为「七、明日展望」

生成链路:
- 修复分段生成未回灌前文,后段看不到前段导致重复铺数据、数字口径打架
- 新增 REQUIRED_TOOLS 校验:核心工具不全时补一轮并在 user 消息点名缺失项
- 修复 _extract_summary 被 # 标题行占掉,改为剥离标题与 Markdown 标记后截断

AGENTS.md 同步八章结构与 prompt 工程约定
2026-09-12 15:57:14 +08:00

503 lines
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""AI 分析核心服务
负责:
1. 调用 OpenAI 兼容 API 进行分析
2. Function Calling 循环(AI 可主动获取数据)
3. 采集当日盘面快照(供次日环比)
4. 保存报告到数据库
"""
import asyncio
import json
import re
import traceback
import httpx
from datetime import datetime, timezone, timedelta
from services.ai_config import AI_API_BASE, AI_API_KEY, AI_MODEL, AI_MAX_TOKENS, AI_TEMPERATURE
from services.ai_tools import TOOLS, execute_tool
from database import get_connection
_CST = timezone(timedelta(hours=8))
SYSTEM_PROMPT = """你是一位资深A股复盘分析师,为收盘后的每日复盘报告供稿。读者是有一定经验、但没时间盯盘的短线与波段投资者。
工作原则:
1. 事实优先:所有数字、股票名称与代码、涨停原因、封单金额必须能在工具返回中找到出处;工具没有的写「—(今日无数据)」,严禁凭记忆、常识或推理补全。
2. 时间锚定:只分析给定交易日当天及之前的真实数据,禁止引用训练记忆中的行情、政策或题材。
3. 判断必须挂数字:每个结论后面要有价格/涨跌幅/金额/家数支撑,不做无数据的定性判断。
4. 说人话:结论先行、短句、少形容词;禁用「整体来看」「值得注意的是」「情绪有所回暖」这类无信息量的套话。
输出格式(前端按 ## 二级标题切卡片渲染,红涨绿跌依赖数字符号):
- 章节标题严格用 `## 一、xxx` 形式,标题不超过 12 字
- 涨跌幅、环比变化、资金净额等有方向的数字必须带 + 或 - 号(如 +2.31%、-15.6亿);不带符号的数字不会被着色
- 多维度对比一律用 Markdown 表格,表头标注单位"""
# 一次工具轮就必须拿全的核心工具:模型常只调 1-2 个就开写,缺失会让对应章节数据空洞
REQUIRED_TOOLS = ("get_market_dashboard", "get_theme_history",
"get_active_core_stocks", "get_news")
DAILY_ANALYSIS_PROMPT = """请对 {trade_date}(A股交易日)做收盘复盘,生成一份完整报告。
■ 第一步 · 取数(必须一次并行完成)
在同一次回复中并行发起以下 4 个工具调用,不要拆成多轮、不要只调其中一部分:
1. get_market_dashboard() —— 指数、涨跌统计、市场温度、连板梯队 limitLadder、板块资金流 sectorFundFlow、两融、海外指数与国内期货 globalMarkets
2. get_theme_history(date="{trade_date}") —— 今日题材涨幅榜
3. get_active_core_stocks() —— 核心股 10 日涨幅矩阵与所属题材
4. get_news(limit=30) —— 今日财经快讯
只有在需要核实某只具体个股时,才额外调用 get_stock_quote / get_fund_flow,合计不超过 2 次。
■ 第二步 · 写作(以下规约对每一部分都生效)
1. 事实约束:所有数字、个股代码与名称、涨停原因、封单金额必须出自工具返回,找不到就写「—(今日无数据)」,禁止推测或用常识填补;禁止用「预计/有望/大概率」代替数据。
2. 环比约束:下方快照中出现过的指标(成交额、涨跌家数、涨停/跌停/炸板、市场温度、两融余额、主要指数),首次提到时必须写成「今日值(较昨日 ±变化)」;快照缺失时写「暂无昨日基准」,不得编造环比。
3. 单位口径(全文统一,表头标注单位):成交额→亿元(≥1万亿时写 x.xx 万亿);资金净额→亿元;封单金额→万元(limitLadder 的 sealWan 字段);指数→点。
4. 符号规范:涨跌幅、环比变化、净流入/流出一律带 + 或 -,这是前端红涨绿跌的着色依据。
5. 连板梯队:limitLadder 中 2 连板及以上全部列出(层级/股票/涨停原因/封单),首板只点评 3-5 只人气股。
6. 板块资金:引用 sectorFundFlow 的行业净流入 TOP3、净流出 TOP3、概念净流入 TOP3(亿元,带符号),并结合题材涨跌说明资金是否印证了盘面。
7. 消息面:基于 get_news 挑 5-8 条对次日影响最大的快讯,每条格式为「【分类】一句话新闻 —— 一句影响解读」(分类限:宏观/政策/行业/公司/海外);快讯里没有的方向如实说明「今日无重要消息」,严禁编造。
8. 不荐股:禁止给出「买入/卖出/加仓/减仓/止损位/几成仓」等指令性表述;机会与风险一律写成条件句(如「若次日××,则××假设被强化/破坏」)。
报告共八章,固定顺序:一、市场总览|二、题材热点分析|三、核心股追踪|四、资金与筹码|五、重要消息面|六、海外市场与国内期货|七、明日展望|八、风险提示。
同一段数据只允许在它归属的那一章出现:指数与量能在第一章、题材涨幅在第二章、连板与核心股在第三章、资金流与两融在第四章,其他章节只做结论引用,不重复铺数据。
系统会把八章拆成三次生成,每次只写被指定的部分。
{prev_snapshot_section}
{prev_report_section}
"""
# 分段生成指令:网关对单次 LLM 请求有约120s硬超时,整篇报告一次生成必被掐断,
# 故拆为三段(每段约1500-2000字),各自独立调用后拼接
REPORT_PARTS = [
"""现在写报告的【第1部分】,只输出这一部分,直接输出 Markdown,禁止「好的/以下是」之类开场白:
1. 以 `# {title} A股收盘分析报告` 一级标题开头;标题后不要写定调摘要、不要写引言,直接进入第一章
2. `## 一、市场总览`:主要指数表格(指数/收盘/涨跌幅)+量能与情绪(成交额、涨跌家数、涨停/跌停/炸板率、市场温度),全部按规约写成「今日值(较昨日 ±变化)」
3. `## 二、题材热点分析`:题材涨幅 TOP5 表(题材/涨幅/龙头/连续活跃天数)、持续活跃与新兴热点、退潮警示
全文控制在 1700 字以内。""",
"""前文(第1部分)已在上下文中,现在续写【第2部分】,直接输出 Markdown,不要复述前文、不要写过渡句:
- `## 三、核心股追踪`:limitLadder 完整连板梯队表(层级/股票/涨停原因/封单万元),2 连板以上全列,首板挑 3-5 只人气股点评;再写核心股 10 日表现与龙头辨识
- `## 四、资金与筹码`:sectorFundFlow 的行业净流入 TOP3 / 净流出 TOP3 / 概念净流入 TOP3(亿元,带符号)、两融余额与变化、事件情报中值得注意的筹码信号
全文控制在 1500 字以内。""",
"""前两部分已在上下文中,现在续写【第3部分】,直接输出 Markdown,不要复述前文、不要写过渡句:
- `## 五、重要消息面`:5-8 条,格式为【分类】一句话新闻 —— 一句影响解读
- `## 六、海外市场与国内期货`:从 globalMarkets 挑与 A 股关联度高的品种(股指期货、原油、贵金属、黑色系),逐一点评对次日的影响;数据缺失如实说明
- `## 七、明日展望`:明日主线方向、需要验证的观察点、节奏与规避方向(用条件句,不给操作指令)
- `## 八、风险提示`:3-5 条,每条写清「什么情况出现 → 说明什么判断失效」
全文控制在 2000 字以内。""",
]
async def _consume_sse(resp: httpx.Response) -> dict:
"""消费 OpenAI 兼容 SSE 流,拼装为与非流式响应相同的结构"""
content_parts: list[str] = []
finish_reason = ""
usage: dict = {}
# tool_calls 按 index 拼装(流式下 arguments 分片到达)
tool_acc: dict[int, dict] = {}
async for line in resp.aiter_lines():
if not line.startswith("data:"):
continue
data = line[5:].strip()
if not data or data == "[DONE]":
continue
try:
chunk = json.loads(data)
except json.JSONDecodeError:
continue
if chunk.get("usage"):
usage = chunk["usage"]
choices = chunk.get("choices") or []
if not choices:
continue
delta = choices[0].get("delta") or {}
if delta.get("content"):
content_parts.append(delta["content"])
for tc in delta.get("tool_calls") or []:
idx = tc.get("index", 0)
slot = tool_acc.setdefault(idx, {"id": "", "type": "function",
"function": {"name": "", "arguments": ""}})
if tc.get("id"):
slot["id"] = tc["id"]
fn = tc.get("function") or {}
if fn.get("name"):
slot["function"]["name"] += fn["name"]
if fn.get("arguments"):
slot["function"]["arguments"] += fn["arguments"]
if choices[0].get("finish_reason"):
finish_reason = choices[0]["finish_reason"]
message: dict = {"role": "assistant", "content": "".join(content_parts) or None}
if tool_acc:
message["tool_calls"] = [
{"id": tool_acc[i]["id"], "type": "function",
"function": tool_acc[i]["function"]}
for i in sorted(tool_acc)
]
return {"choices": [{"message": message, "finish_reason": finish_reason}], "usage": usage}
async def call_llm(messages: list, tools: list = None) -> dict:
"""调用 OpenAI 兼容 API(流式)。
必须用 stream:网关对非流式请求有约120s的代理超时,长生成会被 502 掐断;
流式下字节持续到达不会被判定超时。返回结构与非流式一致。
"""
async with httpx.AsyncClient() as client:
payload = {
"model": AI_MODEL,
"messages": messages,
"stream": True,
}
if AI_MAX_TOKENS is not None:
payload["max_tokens"] = AI_MAX_TOKENS
if AI_TEMPERATURE is not None:
payload["temperature"] = AI_TEMPERATURE
if tools:
payload["tools"] = tools
payload["tool_choice"] = "auto"
max_attempts = 4
for attempt in range(1, max_attempts + 1):
try:
async with client.stream(
"POST",
f"{AI_API_BASE}/chat/completions",
headers={
"Authorization": f"Bearer {AI_API_KEY}",
"Content-Type": "application/json",
"Accept": "text/event-stream",
},
json=payload,
timeout=600,
) as resp:
if resp.status_code == 429 and attempt < max_attempts:
wait = min(30 * attempt, 90)
print(f"[ai-service] LLM 429 限流,{wait}s 后重试(第 {attempt}/{max_attempts - 1} 次)")
await asyncio.sleep(wait)
continue
if resp.status_code >= 500 and attempt < max_attempts:
print(f"[ai-service] LLM {resp.status_code},{min(15 * attempt, 60)}s 后重试")
await asyncio.sleep(min(15 * attempt, 60))
continue
resp.raise_for_status()
return await _consume_sse(resp)
except httpx.TransportError as e:
# 网络层错误(超时/断连)也值得重试
if attempt < max_attempts:
wait = min(15 * attempt, 60)
print(f"[ai-service] LLM 网络错误({type(e).__name__}),{wait}s 后重试")
await asyncio.sleep(wait)
continue
raise
raise RuntimeError("LLM 调用重试次数耗尽")
async def collect_ai_analysis(trade_date: str) -> dict:
"""AI 分析主流程
Args:
trade_date: 交易日 YYYY-MM-DD
Returns:
{"id": int, "tokens_used": int, "tools_used": list}
"""
prev_report_section = _get_prev_report_section(trade_date)
# 采集当日盘面快照(供次日环比),并读取上一交易日快照注入 prompt
await _capture_market_snapshot(trade_date)
prev_snapshot_section = _get_prev_snapshot_section(trade_date)
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": DAILY_ANALYSIS_PROMPT.format(
trade_date=trade_date,
prev_report_section=prev_report_section,
prev_snapshot_section=prev_snapshot_section,
)}
]
# ── 阶段一:工具轮(只获取数据,模型若直接开写报告则丢弃,由阶段二重写) ──
tools_used = []
total_tokens = 0
llm_calls = 0
was_truncated = False
max_rounds = 8
tool_rounds = 0
for i in range(max_rounds):
response = await call_llm(messages, tools=TOOLS)
llm_calls += 1
total_tokens += response.get("usage", {}).get("total_tokens", 0)
choice = response["choices"][0]
message = choice["message"]
finish_reason = choice.get("finish_reason", "")
print(f"[ai-service] tool round {i}: finish={finish_reason}, "
f"content_len={len(message.get('content') or '')}, "
f"tool_calls={len(message.get('tool_calls') or [])}")
if finish_reason == "tool_calls" and message.get("tool_calls"):
tool_rounds += 1
messages.append(message)
for tool_call in message["tool_calls"]:
func_name = tool_call["function"]["name"]
func_args = json.loads(tool_call["function"]["arguments"])
tools_used.append(func_name)
result = await execute_tool(func_name, func_args)
messages.append({
"role": "tool",
"tool_call_id": tool_call["id"],
"content": result
})
# 核心工具齐了就立即进分段写作(再问一轮只会空转120s);
# 不齐则最多再补一轮,否则对应章节会无数据可写
if all(t in tools_used for t in REQUIRED_TOOLS):
break
if tool_rounds >= 2:
print(f"[ai-service] 核心工具仍不齐(已取:{sorted(set(tools_used))}),直接进入写作")
break
print(f"[ai-service] 核心工具不全,补一轮取数({tool_rounds}/2)")
missing = [t for t in REQUIRED_TOOLS if t not in tools_used]
messages.append({
"role": "user",
"content": f"还缺以下数据,请立即调用后停止取数:{', '.join(missing)}"
f"(get_theme_history 的 date 参数为 {trade_date})"
})
continue
# 非工具轮(模型直接开写/空返回):只要有工具结果就直接进入分段写作;
# 一轮工具都没拿到则重试
if tools_used:
break
print(f"[ai-service] 未获取到工具数据,重试({i + 1}/{max_rounds})")
await asyncio.sleep(5)
if not tools_used:
raise RuntimeError("工具数据获取失败,无法生成报告")
# ── 阶段二:分段生成报告(绕开网关单请求约120s硬超时) ──
final_content = ""
for part_idx, part_prompt in enumerate(REPORT_PARTS):
part_content = ""
for attempt, backoff in ((1, 0), (2, 10), (3, 30)):
if backoff:
await asyncio.sleep(backoff)
part_messages = messages + [{
"role": "user",
"content": part_prompt.format(title=trade_date) if part_idx == 0 else part_prompt,
}]
response = await call_llm(part_messages)
llm_calls += 1
total_tokens += response.get("usage", {}).get("total_tokens", 0)
choice = response["choices"][0]
part_content = choice["message"].get("content") or ""
finish = choice.get("finish_reason", "")
print(f"[ai-service] part {part_idx + 1} attempt {attempt}: finish={finish}, len={len(part_content)}")
if part_content and finish in ("stop", "length"):
break
print(f"[ai-service] part {part_idx + 1} 生成异常,重试")
if not part_content:
was_truncated = True
print(f"[ai-service] 警告:part {part_idx + 1} 三次尝试均失败")
part_content = f"\n\n> (第{part_idx + 1}部分生成失败,请稍后重新生成)\n"
else:
# 回灌前文:后段看不到前段内容会导致重复铺数据、数字口径打架
messages.append({"role": "assistant", "content": part_content})
final_content += (final_content and "\n\n" or "") + part_content
summary = _extract_summary(final_content)
report_id = _save_report(trade_date, final_content, summary, tools_used, total_tokens, llm_calls)
print(f"[ai-service] 生成完成:LLM调用 {llm_calls} 次,tokens {total_tokens}")
return {"id": report_id, "tokens_used": total_tokens, "llm_calls": llm_calls, "tools_used": tools_used, "truncated": was_truncated}
def _extract_summary(content: str) -> str:
"""摘要:去掉标题行与 Markdown 标记后截断 200 字(供管理列表展示)
正文首行是 `# XXXX A股收盘分析报告`,直接截断会让它占掉摘要大半。
"""
text = re.sub(r"^#.*$", "", content or "", flags=re.MULTILINE)
text = re.sub(r"[>#*`|]", "", text)
return re.sub(r"\s+", " ", text).strip()[:200]
def _num(v) -> str:
"""快照数值安全转字符串"""
if isinstance(v, float):
return f"{v:g}"
return str(v if v is not None else "-")
def _fmt_amount(v) -> str:
"""成交额(元)→ 万亿/亿 可读格式"""
try:
v = float(v)
except (TypeError, ValueError):
return "-"
if v >= 1e12:
return f"{v / 1e12:.2f}万亿"
if v >= 1e8:
return f"{v / 1e8:.0f}亿"
return f"{v:.0f}元"
def _fmt_index(idx: dict) -> str:
if not idx:
return "-"
try:
pct = round(float(idx.get("changePct", 0)), 2)
except (TypeError, ValueError):
pct = 0
sign = "+" if pct >= 0 else ""
return f"{idx.get('price', '-')}({sign}{pct}%)"
def _fmt_temperature(v) -> str:
"""温度可能是 dict(score/label/factors),取分数与标签"""
if isinstance(v, dict):
score = v.get("score")
label = v.get("label") or ""
return f"{score}分{('(' + label + ')') if label else ''}"
return _num(v)
async def _capture_market_snapshot(trade_date: str) -> bool:
"""采集当日盘面快照入库(指数+涨跌统计),供次日分析做环比"""
try:
from routes.market_dashboard import _build_dashboard
data = await _build_dashboard()
stats = data.get("marketStats") or {}
# 数据有效性校验:fuyao 拉取失败时涨跌统计全 0,空快照会污染次日环比
if not data.get("indices") or (stats.get("upCount", 0) + stats.get("downCount", 0) == 0):
print(f"[ai-service] 盘面数据无效,跳过快照入库 {trade_date}")
return False
payload = json.dumps(
{"indices": data.get("indices", []), "marketStats": stats},
ensure_ascii=False,
default=str,
)
conn = get_connection()
try:
conn.execute(
"""INSERT INTO daily_market_stats (trade_date, payload) VALUES (?, ?)
ON CONFLICT(trade_date) DO UPDATE SET payload = excluded.payload""",
(trade_date, payload),
)
conn.commit()
finally:
conn.close()
return True
except Exception:
print(f"[ai-service] 市场快照采集失败 {trade_date}:")
traceback.print_exc()
return False
def _get_prev_snapshot_section(trade_date: str) -> str:
"""读取上一交易日盘面快照,格式化为 prompt 中的环比数据段"""
conn = get_connection()
try:
row = conn.execute(
"SELECT trade_date, payload FROM daily_market_stats WHERE trade_date < ? ORDER BY trade_date DESC LIMIT 1",
(trade_date,),
).fetchone()
if not row:
return ""
try:
snap = json.loads(row["payload"])
except (TypeError, ValueError):
return ""
stats = snap.get("marketStats") or {}
indices = {i.get("name"): i for i in (snap.get("indices") or []) if isinstance(i, dict)}
idx_line = "、".join(
f"{name} {_fmt_index(indices.get(name))}"
for name in ("上证指数", "深证成指", "创业板指", "科创50")
)
margin_line = ""
if stats.get("marginBalanceYi"):
change = stats.get("marginChangeYi")
change_txt = ""
if change is not None:
sign = "+" if float(change) >= 0 else ""
change_txt = f"(较前一日 {sign}{_num(change)}亿)"
margin_line = f"\n- 两融余额:{_num(stats.get('marginBalanceYi'))}亿{change_txt},数据日期 {stats.get('marginDate') or '-'}(T+1)"
return f"""以下是前一交易日({row["trade_date"]})的盘面数据快照,报告中的量能与情绪数字必须给出与它的环比对比:
- 两市成交额:{_fmt_amount(stats.get("totalTurnover"))}
- 上涨/下跌/平盘:{_num(stats.get("upCount"))}/{_num(stats.get("downCount"))}/{_num(stats.get("flatCount"))},涨停 {_num(stats.get("limitUp"))} 家、跌停 {_num(stats.get("limitDown"))} 家、炸板 {_num(stats.get("limitBreak"))} 家(炸板率 {_num(stats.get("breakRate"))}%)
- 强势/弱势股:{_num(stats.get("strongCount"))}/{_num(stats.get("weakCount"))},市场宽度 {_num(stats.get("marketBreadth"))}%
- 市场温度:{_fmt_temperature(stats.get("temperature"))},竞价信号:{stats.get("auctionSignal") or "-"}{margin_line}
- 指数收盘:{idx_line}
---
"""
finally:
conn.close()
def _save_report(trade_date: str, content: str, summary: str, tools_used: list, tokens_used: int, llm_calls: int = 0) -> int:
"""保存报告到数据库(同日重生成:覆盖内容、generation_count+1、tokens 记当次消耗)"""
conn = get_connection()
try:
conn.execute(
"""INSERT INTO ai_reports
(trade_date, report_type, title, content, summary, tools_used, model, tokens_used, llm_calls, updated_at)
VALUES (?, 'daily', ?, ?, ?, ?, ?, ?, ?, datetime('now','localtime'))
ON CONFLICT(trade_date, report_type) DO UPDATE SET
title = excluded.title,
content = excluded.content,
summary = excluded.summary,
tools_used = excluded.tools_used,
model = excluded.model,
tokens_used = excluded.tokens_used,
llm_calls = excluded.llm_calls,
updated_at = excluded.updated_at,
generation_count = ai_reports.generation_count + 1""",
(
trade_date,
f"{trade_date} A股收盘分析",
content,
summary,
json.dumps(tools_used),
AI_MODEL,
tokens_used,
llm_calls,
),
)
conn.commit()
row = conn.execute(
"SELECT id FROM ai_reports WHERE trade_date = ? AND report_type = 'daily'",
(trade_date,),
).fetchone()
return row["id"]
finally:
conn.close()
def _get_prev_report_section(trade_date: str) -> str:
"""获取前一个交易日的报告,用于上下文参考"""
conn = get_connection()
try:
row = conn.execute(
"SELECT trade_date, content FROM ai_reports WHERE trade_date < ? AND report_type = 'daily' ORDER BY trade_date DESC LIMIT 1",
(trade_date,)
).fetchone()
if not row:
return ""
prev_date = row["trade_date"]
prev_content = row["content"] or ""
return f"""以下是前一个交易日({prev_date})的分析报告,仅用于保持分析连续性:
- 今日数据优先:两者冲突时以今日工具数据为准,并在正文点出变化(如"昨日提示的××今日××")
- 禁止照搬昨日结论、禁止复制昨日段落
- 重点核对昨日"明日展望"与"风险提示"里提到的方向是否兑现;未兑现的要在正文中说明
{prev_content}
---
"""
finally:
conn.close()