Skip to content

Commit afca35f

Browse files
committed
feat(ai): 切片并发 2-3 + 单片重试 + 部分成功——任务执行拆至 ai_refine_task (F2-B4)
1 parent 6e72c5d commit afca35f

6 files changed

Lines changed: 348 additions & 224 deletions

File tree

‎app/src-tauri/src/ai_cost.rs‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,13 +16,17 @@
1616
use std::collections::HashMap;
1717

1818
/// 默认单价(元/百万 token;免费档 0——2026-08 选型 R1-0528-Qwen3-8B ¥0/M)。
19+
/// 保留兼容(测试断言 + 无模型回退语义),生产路径走 price_for_model。
20+
#[allow(dead_code)] // 兼容 API:测试断言 + 未来单价表整体覆盖入口
1921
pub const DEFAULT_PRICE_PER_1M: f64 = 0.0;
2022
/// env 覆盖键(元/百万 token;整体覆盖映射表——开发路径)。
2123
const PRICE_ENV_KEY: &str = "SILICONFLOW_PRICE_PER_1M_TOKENS";
2224
/// 输出 token 估算系数(输入→输出比例;精修/补充是重写型任务,输出量
2325
/// 接近输入量——1.0 保守上界;实测校准随 golden 冒烟)。
2426
const OUTPUT_RATIO: f64 = 1.0;
25-
/// 未知模型警告文案(确认弹窗可见——成本透明铁律)。
27+
/// 未知模型警告文案(确认弹窗展示——成本透明铁律;前端经 priceKnown
28+
/// 字段自行渲染,本常量保留为文案单一来源备查)。
29+
#[allow(dead_code)] // 文案单一来源(前端内联同文案;未来富化时消费)
2630
const UNKNOWN_MODEL_WARN: &str = "(该模型单价未登记,费用可能不准确)";
2731

2832
/// 内置单价映射表(模型名 → 元/百万 token)。
@@ -75,6 +79,7 @@ pub fn estimate_cost(tokens: usize, price_per_1m: f64) -> f64 {
7579
/// 单价解析(env 覆盖;缺省/非法 → 默认 0——免费档兜底)。
7680
///
7781
/// @ai-context: 保留向后兼容(旧调用方);新代码走 price_for_model。
82+
#[allow(dead_code)] // 兼容 API:usage_cost 沿用 + 测试断言
7883
pub fn price_per_1m() -> f64 {
7984
price_for_model("").0
8085
}
@@ -108,6 +113,7 @@ pub fn estimate_for_content_model(chars: usize, model: &str) -> CostEstimate {
108113
}
109114

110115
/// 按内容字符数估算(兼容旧签名——免费档默认模型,未知模型警告保留)。
116+
#[allow(dead_code)] // 兼容 API:测试 + 旧调用方(新代码走 _model 版)
111117
pub fn estimate_for_content(chars: usize) -> CostEstimate {
112118
estimate_for_content_model(chars, "")
113119
}
@@ -119,6 +125,7 @@ pub fn usage_cost(tokens_in: usize, tokens_out: usize) -> f64 {
119125
}
120126

121127
/// 未知模型警告文案(确认弹窗拼接用)。
128+
#[allow(dead_code)] // 文案单一来源(前端经 priceKnown 自行渲染同文案)
122129
pub fn unknown_model_warning() -> &'static str {
123130
UNKNOWN_MODEL_WARN
124131
}
Lines changed: 329 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,329 @@
1+
//! AI 精修任务执行(F2-B4 拆分,2026-08-21:自 commands_ai_refine.rs 拆出——
2+
//! 豁免清单拆分计划兑现;行数回归 ≤300)。
3+
//!
4+
//! @ai-context: 后台任务主体:规则草稿 → 切片 → 并发精修(F2:并发 2-3,
5+
//! 单片失败重试 1 次,仍失败保留已成功片 = 部分成功,不再
6+
//! 一片失败全任务失败)→ 合并 → diff。panic 由 catch_unwind
7+
//! 归一 Failed(状态流转永不失联)。审计/任务落库在此层完成。
8+
//! @ai-context: 并发实现:std::thread::scope + channel 工作池(spawn_blocking
9+
//! 线程内再开 worker——网络调用阻塞 worker 线程,互不干扰;
10+
//! AiClient 为 Clone+Send,跨线程共享安全)。
11+
12+
use std::collections::VecDeque;
13+
use std::sync::mpsc;
14+
use std::sync::{Arc, Mutex};
15+
16+
use crate::ai_client::AiClient;
17+
use crate::ai_mock::AiMockAdapter;
18+
use crate::ai_note_refine::AiNoteRefineAdapter;
19+
use crate::ai_refine_protocol::AiRefineRequest;
20+
use crate::ai_task::{slice_note, AiTaskFailure, AiTaskState, SLICE_MAX_CHARS};
21+
use crate::commands::AppState;
22+
use crate::commands_ai_refine::{set_task, AiRefineResult};
23+
use crate::commands_session_note::build_rule_draft_with_analysis;
24+
use crate::note_diff::{diff_markdown, diff_stats};
25+
use crate::note_filter::PurifyEnv;
26+
use crate::outline::{detect_outline_smart, OutlineConfig};
27+
use crate::video_profile::ProfileKind;
28+
29+
/// 切片并发上限(REQ-145:并发 2-3——配额并发安全由 command 层启动前按
30+
/// 预估片数一次性消耗保证,此处 worker 数不超切片数)。
31+
const CONCURRENCY: usize = 3;
32+
/// 单片失败重试次数(幂等片——同片重跑不产生副作用)。
33+
const SLICE_RETRY: usize = 1;
34+
35+
/// 精修任务审计记录(F1:REQ-140 轨迹可见化——summary 不含原文,隐私红线)。
36+
fn push_refine_audit(st: &AppState, session_id: i64, result: &str, model: Option<&str>) {
37+
let now = crate::db_sessions_rows::unix_seconds();
38+
if let Ok(mut g) = st.ai_guardrails.lock() {
39+
g.push_audit(crate::ai_guardrails::AiAuditEntry {
40+
at_unix: now,
41+
upload_summary: format!(
42+
"refine session={} model={}",
43+
session_id,
44+
model.unwrap_or("?")
45+
),
46+
result: result.to_string(),
47+
});
48+
}
49+
}
50+
51+
/// 后台精修任务:规则草稿 → 切片 → 逐片精修(mock/云端)→ 合并 → diff。
52+
///
53+
/// @ai-context: 彻底检测加固(2026-08-21):spawn_blocking 的 JoinHandle 未被
54+
/// await——闭包内 panic 会被 tokio 吞掉,任务状态永久停在
55+
/// Pending(前端永久显示"任务排队中",无失败可重试)。
56+
/// catch_unwind 把 panic 归一为 Failed 状态,状态流转永不失联。
57+
/// @ai-context: F2-B4:单片失败重试后仍失败 → 保留已成功片(partial_failed
58+
/// 语义:failed_slices > 0,前端显示"部分成功 x/y 片")。
59+
pub fn run_refine_task(st: AppState, task_id: u64, session_id: i64, mock: bool) {
60+
// 诊断日志(2026-08-21 真机"排队中"排查):tauri dev 终端可见各阶段进度
61+
eprintln!("[refine-task] task={} start session={} mock={}", task_id, session_id, mock);
62+
let started = std::time::Instant::now();
63+
let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
64+
run_refine_task_inner(&st, task_id, session_id, mock)
65+
}))
66+
.unwrap_or_else(|_| {
67+
Err(AiTaskFailure::Other(
68+
"精修任务内部错误(panic)——请重试;若复现请反馈".to_string(),
69+
))
70+
});
71+
let elapsed_ms = started.elapsed().as_millis() as i64;
72+
match outcome {
73+
Ok(result) => {
74+
eprintln!(
75+
"[refine-task] task={} succeeded slices={} failed={} diff={}",
76+
task_id,
77+
result.slices,
78+
result.failed_slices,
79+
result.diff.len()
80+
);
81+
{
82+
let mut tasks = st.ai_tasks.lock().unwrap_or_else(|e| e.into_inner());
83+
if let Some(entry) = tasks.get_mut(&task_id) {
84+
entry.result = serde_json::to_value(&result).ok();
85+
}
86+
}
87+
set_task(&st, task_id, AiTaskState::Succeeded);
88+
// F1 修复(2026-08-21):精修调用上审计——REQ-140 轨迹可见化
89+
// (此前只有余额/测试连接/复核有记录,精修补充零审计)
90+
push_refine_audit(&st, session_id, "ok", Some(&result.model));
91+
// F2 任务中心:终态落库(写库失败不阻断——H2 设计)
92+
let result_json = serde_json::to_string(&result).ok();
93+
let _ = st.db.finish_ai_task(
94+
task_id,
95+
"succeeded",
96+
result_json.as_deref(),
97+
None,
98+
elapsed_ms,
99+
);
100+
}
101+
Err(reason) => {
102+
// 打印具体 message——区分"未配置密钥"vs"密钥无效(401/403)"(真机排查)
103+
eprintln!(
104+
"[refine-task] task={} failed kind={} msg={}",
105+
task_id,
106+
reason.kind(),
107+
reason.message()
108+
);
109+
set_task(&st, task_id, AiTaskState::Failed { reason: reason.clone() });
110+
push_refine_audit(&st, session_id, "error", None);
111+
let _ = st.db.finish_ai_task(
112+
task_id,
113+
"failed",
114+
None,
115+
Some(&format!("{}: {}", reason.kind(), reason.message())),
116+
elapsed_ms,
117+
);
118+
}
119+
}
120+
}
121+
122+
/// 精修任务主体(返回 Result;panic 由外层 catch_unwind 兜底)。
123+
fn run_refine_task_inner(
124+
st: &AppState,
125+
task_id: u64,
126+
session_id: i64,
127+
mock: bool,
128+
) -> Result<AiRefineResult, AiTaskFailure> {
129+
let env = PurifyEnv {
130+
config: st.purify.clone(),
131+
symbol: st.symbol_normalize.clone(),
132+
corrections: st.ocr_corrections.clone(),
133+
};
134+
// ① 规则草稿 + 结构分析一次完成(审查修复 2026-08-21:build_rule_draft_
135+
// with_analysis 返回 analysis——章节/术语直接复用,消除二次 analyze 双跑)
136+
eprintln!("[refine-task] task={} 阶段①构建规则草稿(本地分析)", task_id);
137+
let (draft, analysis) =
138+
build_rule_draft_with_analysis(&st.db, &st.ui_junk, &env, &st.data_dir, session_id, None)
139+
.map_err(AiTaskFailure::Other)?;
140+
eprintln!("[refine-task] task={} 草稿完成 markdown={} 字符", task_id, draft.markdown.chars().count());
141+
// ② 精修上下文(档案/章节/术语——analysis 已含章节边界与术语表)
142+
let session = st
143+
.db
144+
.get_session(session_id)
145+
.map_err(|e| AiTaskFailure::Other(e.to_string()))?
146+
.ok_or_else(|| AiTaskFailure::Other("会话不存在".to_string()))?;
147+
let kind = session
148+
.profile
149+
.as_deref()
150+
.map(ProfileKind::parse)
151+
.unwrap_or(ProfileKind::Lecture);
152+
let ocr_blocks = st
153+
.db
154+
.list_ocr_blocks(session_id)
155+
.map_err(|e| AiTaskFailure::Other(e.to_string()))?;
156+
let outline = detect_outline_smart(&ocr_blocks, &draft.ocr_screens, &OutlineConfig::default());
157+
let chapters: Vec<String> = if outline.is_empty() {
158+
analysis
159+
.chapters
160+
.iter()
161+
.enumerate()
162+
.map(|(i, _)| format!("第 {} 节", i + 1))
163+
.collect()
164+
} else {
165+
outline.iter().map(|e| e.text.clone()).collect()
166+
};
167+
let glossary: Vec<String> = analysis.glossary.iter().map(|g| g.term.clone()).collect();
168+
// ③ 切片(≤8000 字/片;进度按片上报)
169+
let slices = slice_note(&draft.markdown, SLICE_MAX_CHARS);
170+
let total = slices.len();
171+
eprintln!("[refine-task] task={} 切片 {} 片", task_id, total);
172+
set_task(st, task_id, AiTaskState::Running { finished_slices: 0, total_slices: total });
173+
let settings = st
174+
.ai_settings
175+
.lock()
176+
.map_err(|e| AiTaskFailure::Other(e.to_string()))?
177+
.clone();
178+
let env_key = std::env::var("SILICONFLOW_API_KEY").ok().filter(|k| !k.is_empty());
179+
let stored_key = st.ai_credentials.load_key().ok().flatten();
180+
// 密钥来源诊断(脱敏:只打长度+前 6 字符;真机 unauthorized 排查 2026-08-21)
181+
eprintln!(
182+
"[refine-task] task={} key: env={} stored={}",
183+
task_id,
184+
env_key
185+
.as_ref()
186+
.map(|k| format!("{}:{}..", k.len(), &k[..6.min(k.len())]))
187+
.unwrap_or_else(|| "无".to_string()),
188+
stored_key
189+
.as_ref()
190+
.map(|k| format!("{}:{}..", k.len(), &k[..6.min(k.len())]))
191+
.unwrap_or_else(|| "无".to_string()),
192+
);
193+
let client = AiClient::from_settings(&settings, stored_key);
194+
let adapter = AiNoteRefineAdapter::new(client.clone());
195+
let mock_adapter = AiMockAdapter;
196+
// F2-B4:并发精修(worker 池消费切片队列;按片上报进度;失败片重试后
197+
// 仍失败 → 记 failed 下标,不中断其他片——部分成功语义)
198+
let (markdowns, failed) = refine_slices_concurrent(RefineCtx {
199+
slices: &slices,
200+
chapters: &chapters,
201+
glossary: &glossary,
202+
profile: kind.as_str(),
203+
adapter: &adapter,
204+
mock_adapter: &mock_adapter,
205+
mock,
206+
workers: total.min(CONCURRENCY),
207+
st,
208+
task_id,
209+
});
210+
let mut refined = markdowns.join("\n\n");
211+
// ④ 合并 + 与规则版 diff(基线=本地版,AI 变化点高亮)
212+
// 丢图修复(2026-08-21 F1):协议 v2 前,模型可能丢弃规则版画面配图行
213+
// (`- ![画面 N](session-images/..)`)——本地合并降级:AI 未保留配图时
214+
// 把规则版配图行按章节合并回精修版(不丢不假,零模型成本)
215+
refined = crate::note_image_merge::merge_rule_images(&draft.markdown, &refined);
216+
let diff = diff_markdown(&draft.markdown, &refined);
217+
let (added, removed, _) = diff_stats(&diff);
218+
Ok(AiRefineResult {
219+
title: draft.title.clone(),
220+
base_markdown: draft.markdown.clone(),
221+
refined_markdown: refined,
222+
diff,
223+
added_lines: added,
224+
removed_lines: removed,
225+
slices: total,
226+
failed_slices: failed,
227+
model: client.config.model,
228+
})
229+
}
230+
231+
/// 并发精修上下文(参数聚合——clippy too_many_arguments 修复)。
232+
struct RefineCtx<'a> {
233+
slices: &'a [String],
234+
chapters: &'a [String],
235+
glossary: &'a [String],
236+
profile: &'a str,
237+
adapter: &'a AiNoteRefineAdapter,
238+
mock_adapter: &'a AiMockAdapter,
239+
mock: bool,
240+
workers: usize,
241+
st: &'a AppState,
242+
task_id: u64,
243+
}
244+
245+
/// 并发切片精修(纯编排):worker 池从共享队列取片 → 单片重试 → 收集。
246+
///
247+
/// @ai-context: 返回 (各片 markdown(保序,失败片跳过), 失败片数)。
248+
/// 单片失败不 panic、不中断其他片——部分成功语义(REQ-145)。
249+
/// 进度经 set_task 上报(finished = 已完成的片数,含失败片——
250+
/// 前端进度条推进不受单片失败影响)。
251+
fn refine_slices_concurrent(ctx: RefineCtx<'_>) -> (Vec<String>, usize) {
252+
let total = ctx.slices.len();
253+
if total == 0 {
254+
return (Vec::new(), 0);
255+
}
256+
let queue: Arc<Mutex<VecDeque<usize>>> = Arc::new(Mutex::new((0..total).collect()));
257+
let (tx, rx) = mpsc::channel::<(usize, Option<String>)>();
258+
let workers = ctx.workers.max(1).min(total);
259+
// 请求一次构建(Arc 共享——worker 只读,避免每 worker 重复克隆切片)
260+
let reqs: Arc<Vec<AiRefineRequest>> = Arc::new(
261+
ctx.slices
262+
.iter()
263+
.map(|s| AiRefineRequest {
264+
content: s.clone(),
265+
profile: ctx.profile.to_string(),
266+
glossary: ctx.glossary.to_vec(),
267+
chapters: ctx.chapters.to_vec(),
268+
})
269+
.collect(),
270+
);
271+
std::thread::scope(|scope| {
272+
for _ in 0..workers {
273+
let queue = queue.clone();
274+
let tx = tx.clone();
275+
let reqs = reqs.clone();
276+
// worker 捕获 ctx 字段(adapter/mock 只读共享;task_id 复制)
277+
let adapter = ctx.adapter;
278+
let mock_adapter = ctx.mock_adapter;
279+
let mock = ctx.mock;
280+
let task_id = ctx.task_id;
281+
scope.spawn(move || loop {
282+
let idx = {
283+
let mut q = queue.lock().unwrap_or_else(|e| e.into_inner());
284+
q.pop_front()
285+
};
286+
let Some(idx) = idx else { break };
287+
let req = &reqs[idx];
288+
let mut outcome: Option<String> = None;
289+
for attempt in 0..=SLICE_RETRY {
290+
let resp = if mock {
291+
Ok(mock_adapter.refine(req))
292+
} else {
293+
adapter.refine(req).map_err(AiTaskFailure::from)
294+
};
295+
match resp {
296+
Ok(r) => {
297+
outcome = Some(r.to_markdown());
298+
break;
299+
}
300+
Err(e) if attempt < SLICE_RETRY => {
301+
eprintln!("[refine-task] task={} 片 {} 第{}次失败,重试: {}", task_id, idx + 1, attempt + 1, e.message());
302+
}
303+
Err(e) => {
304+
eprintln!("[refine-task] task={} 片 {} 重试后仍失败(保留已成功片): {}", task_id, idx + 1, e.message());
305+
break;
306+
}
307+
}
308+
}
309+
let _ = tx.send((idx, outcome));
310+
});
311+
}
312+
drop(tx); // 所有 worker 结束后关闭通道(scope 内最后一个持有者)
313+
});
314+
// 收集(按消息携带的真实切片下标落位——channel 到达序 ≠ 切片序);
315+
// 进度按已收片数上报(含失败片——进度条推进不受单片失败影响)
316+
let received: Vec<(usize, Option<String>)> = rx.iter().collect();
317+
let mut by_index: Vec<Option<String>> = vec![None; total];
318+
for (pos, (slice_idx, out)) in received.iter().enumerate() {
319+
by_index[*slice_idx] = out.clone();
320+
set_task(
321+
ctx.st,
322+
ctx.task_id,
323+
AiTaskState::Running { finished_slices: pos + 1, total_slices: total },
324+
);
325+
}
326+
let failed = by_index.iter().filter(|o| o.is_none()).count();
327+
let markdowns: Vec<String> = by_index.into_iter().flatten().collect();
328+
(markdowns, failed)
329+
}

‎app/src-tauri/src/app_setup.rs‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -156,6 +156,8 @@ pub fn setup_app_state(app: &mut tauri::App) -> Result<(), String> {
156156
// v0.8.0 F2(2026-08-21):任务中心——启动恢复未采纳的成功结果
157157
// (重启不丢;注册表 + id 序列以恢复结果为基准,防 id 冲突覆盖)
158158
{
159+
// 保留策略先行(清理超限旧终态——防表膨胀)
160+
let _ = db.trim_ai_tasks();
159161
let restored = db
160162
.list_restorable_succeeded(100)
161163
.unwrap_or_else(|e| {

0 commit comments

Comments
 (0)