batch_text: 多行文本,每行一个提示词
(batch_text, sys_prompt, temperature, max_new_tokens, device,
model_path, device_map, torch_dtype, state)
| 159 | return "", f"推理失败:{e}", state |
| 160 | |
| 161 | def run_batch(batch_text, sys_prompt, temperature, max_new_tokens, device, |
| 162 | model_path, device_map, torch_dtype, state): |
| 163 | """ |
| 164 | batch_text: 多行文本,每行一个提示词 |
| 165 | """ |
| 166 | if not batch_text or not str(batch_text).strip(): |
| 167 | return "", "请在左侧输入区每行填写一个提示词。", state |
| 168 | |
| 169 | lines = [l.strip() for l in batch_text.splitlines() if l.strip()] |
| 170 | if not lines: |
| 171 | return "", "未检测到有效提示词。", state |
| 172 | |
| 173 | state = ensure_enhancer(state, model_path, device_map, torch_dtype) |
| 174 | enhancer = state["enhancer"] |
| 175 | |
| 176 | results = [] |
| 177 | total_t0 = time.time() |
| 178 | for i, line in enumerate(lines, 1): |
| 179 | t0 = time.time() |
| 180 | try: |
| 181 | out = enhancer.predict( |
| 182 | prompt_cot=line, |
| 183 | sys_prompt=sys_prompt, |
| 184 | temperature=temperature, |
| 185 | max_new_tokens=max_new_tokens, |
| 186 | device=device |
| 187 | ) |
| 188 | dt = time.time() - t0 |
| 189 | results.append(f"[{i}] 原始: {line}\n 重写: {out}\n 耗时: {dt:.2f}s\n") |
| 190 | except Exception as e: |
| 191 | results.append(f"[{i}] 原始: {line}\n 失败: {e}\n") |
| 192 | total_dt = time.time() - total_t0 |
| 193 | summary = f"共处理 {len(lines)} 条,累计耗时:{total_dt:.2f}s" |
| 194 | return "\n".join(results), summary, state |
| 195 | |
| 196 | |
| 197 | # 示例数据 |
nothing calls this directly
no test coverage detected