diff --git a/.env b/.env index 96a0cc4..3386f4b 100644 --- a/.env +++ b/.env @@ -9,9 +9,14 @@ REDIS_URI=redis://localhost:6379/0 # LLM_BASE_URL="http://192.168.137.96:8000/v1" # LLM_MODEL = "Qwen/Qwen3.6-35B-A3B-FP8" # ASR_TTS_BASE_URL = "http://192.168.137.96:8100" -ASR_BASE_URL = "http://192.168.137.96:8901/v1" -TTS_BASE_URL = "http://192.168.137.96:8902" -RAG_BASE_URL = "http://192.168.137.96:8903" +# ASR_BASE_URL = "http://192.168.137.96:8901/v1" +# TTS_BASE_URL = "http://192.168.137.96:8902" +# RAG_BASE_URL = "http://192.168.137.96:8903" + + +ASR_BASE_URL = "http://192.168.137.164:8901/v1" +TTS_BASE_URL = "http://192.168.137.164:8902" +RAG_BASE_URL = "http://192.168.137.164:8903" #6000pro： # LLM_API_KEY='semi-ai.tech' diff --git a/README.md b/README.md index b663539..dc136c3 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # Agent Web Chat -基于 LangGraph 的多模态对话 Agent，支持文本、图片、音频输入，具备屏幕截图、浏览器控制、RAG 知识库、数据库查询等工具调用能力。 +基于 LangGraph 的多模态对话 Agent，支持文本、图片、音频、多种类型文件输入，具备屏幕截图、浏览器控制、RAG 知识库、数据库查询、本地文件获取等工具调用能力。 ## 快速开始 diff --git a/llm_server/rag_server.py b/llm_server/rag_server.py index feec6dc..00dff03 100644 --- a/llm_server/rag_server.py +++ b/llm_server/rag_server.py @@ -71,7 +71,7 @@ def load_models(): global token_true_id, token_false_id, prefix_tokens, suffix_tokens, max_rerank_length token_false_id = rerank_tokenizer.convert_tokens_to_ids("no") token_true_id = rerank_tokenizer.convert_tokens_to_ids("yes") - max_rerank_length = 8192 + max_rerank_length = 32768 prefix = "<|im_start|>system\nJudge whether the Document meets the requirements based on the Query and the Instruct provided. Note that the answer can only be \"yes\" or \"no\".<|im_end|>\n<|im_start|>user\n" suffix = "<|im_end|>\n<|im_start|>assistant\n\n\n\n\n" @@ -111,7 +111,6 @@ async def get_embeddings(req: EmbedRequest): input_texts, padding=True, truncation=True, - max_length=8192, return_tensors="pt", ).to(embed_model.device) diff --git a/llm_server/start_tts.sh b/llm_server/start_tts.sh index 49ca8c1..a1e58b5 100644 --- a/llm_server/start_tts.sh +++ b/llm_server/start_tts.sh @@ -3,7 +3,7 @@ set -euo pipefail # 自动获取脚本所在目录 SCRIPT_DIR="$(cd -- "$(dirname "${BASH_SOURCE[0]}")" && pwd)" -VENV_DIR="${SCRIPT_DIR}/.venv_audio" +VENV_DIR="${SCRIPT_DIR}/myenv" HOST_PORT="${HOST_PORT:-8902}" # 激活虚拟环境 @@ -23,13 +23,12 @@ cd "${SCRIPT_DIR}" # 设置环境变量（和 vLLM 保持一致） export MODELSCOPE_DOWNLOAD_MODE=local export MODELSCOPE_HUB_ENABLE_FILE_CHECK=false -export MODELSCOPE_CACHE="${SCRIPT_DIR}/modelscope" +export MODELSCOPE_CACHE="${SCRIPT_DIR}/models" export HF_HUB_OFFLINE=1 export TRANSFORMERS_OFFLINE=1 -# 启动 FastAPI 服务 +# 启动 FastAPI 服务，使用 4 个 worker 进程 uvicorn tts_server:app \ --host 0.0.0.0 \ --port "${HOST_PORT}" \ - --workers 1 \ - + --workers 4 diff --git a/llm_server/test_tts.py b/llm_server/test_tts.py index 38b28d2..54d5b78 100644 --- a/llm_server/test_tts.py +++ b/llm_server/test_tts.py @@ -1,71 +1,153 @@ -import requests -import argparse -import os +#!/usr/bin/env python3 +""" +真实 TTS API 测试（增强版）：验证分块合并 + 并发耗时。 +""" + +import io +import wave +import re import time +import os +import sys +from concurrent.futures import ThreadPoolExecutor, as_completed +import requests -# 默认服务地址 -BASE_URL = "http://127.0.0.1:8000" +# ======================== 配置 ======================== +TTS_BASE_URL = "http://192.168.137.164:8902" +OUTPUT_WAV = "test_output.wav" +REQUEST_TIMEOUT = 120 +MAX_PARALLEL = 4 +MIN_EFFECTIVE_CHARS = 20 +# ===================================================== -def test_health(): - """检查服务是否在线""" - resp = requests.get(f"{BASE_URL}/health") - print(f"Health check: {resp.status_code} {resp.json()}") +SENT_SPLIT_PATTERN = re.compile(r'(?<=[。！？；，、])') +REMOVE_PUNCT_PATTERN = re.compile(r'[。！？；，、：；“”（）《》【】\-\_\—\~\.\,\!\?\s·]') -def test_tts(text, speaker=None, speed=1.0, output="output.wav"): - """ - 调用 TTS 接口并保存音频文件 - """ - params = {"text": text, "speed": speed} - if speaker: - params["speaker"] = speaker +def effective_length(text: str) -> int: + return len(REMOVE_PUNCT_PATTERN.sub('', text)) - print(f"Requesting TTS...") - start = time.time() - resp = requests.get(f"{BASE_URL}/tts", params=params) - elapsed = time.time() - start +def split_text(text: str, min_effective_chars: int = MIN_EFFECTIVE_CHARS) -> list[str]: + raw_chunks = [c.strip() for c in SENT_SPLIT_PATTERN.split(text) if c.strip()] + if not raw_chunks: + return [text] - if resp.status_code == 200: - with open(output, "wb") as f: - f.write(resp.content) - print(f"Saved audio to: {output}") - print(f"Audio size: {os.path.getsize(output)} bytes") - print(f"Time taken: {elapsed:.2f}s") - else: - print(f"Error {resp.status_code}: {resp.text}") - -def test_speakers(): - """尝试获取可用说话人列表（通过 health 端点里可能不返回，这里直接尝试几个常见说话人）""" - # MeloTTS 中文常见说话人: ZH, ZH_CN, ZH_TW 等，我们可以列出文档或手动尝试 - test_speakers = ["ZH", "ZH_CN", "ZH_TW", "EN", "JP"] - print("Testing available speakers...") - for spk in test_speakers: - params = {"text": "测试", "speaker": spk} - resp = requests.get(f"{BASE_URL}/tts", params=params) - if resp.status_code == 200: - print(f" {spk}: OK") + merged = [] + buffer = "" + for chunk in raw_chunks: + buffer += chunk + if effective_length(buffer) >= min_effective_chars: + merged.append(buffer) + buffer = "" + if buffer: + if merged: + merged[-1] += buffer else: - print(f" {spk}: {resp.text[:50]}") + merged.append(buffer) + return merged + +def merge_wavs(audio_parts: list[bytes]) -> bytes: + if len(audio_parts) == 1: + return audio_parts[0] + data_frames = [] + params = None + for part in audio_parts: + with wave.open(io.BytesIO(part), 'rb') as wf: + if params is None: + params = wf.getparams() + data_frames.append(wf.readframes(wf.getnframes())) + buf = io.BytesIO() + with wave.open(buf, 'wb') as wf_out: + wf_out.setparams(params) + for frames in data_frames: + wf_out.writeframes(frames) + return buf.getvalue() + +def wav_duration(audio_bytes: bytes) -> float: + with wave.open(io.BytesIO(audio_bytes), 'rb') as wf: + return wf.getnframes() / wf.getframerate() if wf.getframerate() > 0 else 0.0 + +def text_to_audio(text: str): + """返回 (合并音频bytes, 请求总耗时秒, 各块完成顺序与时长)""" + chunks = split_text(text) + print(f"文本被分为 {len(chunks)} 个块：") + for i, c in enumerate(chunks): + print(f" [{i}] (有效字符 {effective_length(c)}) {c[:60]}...") + + if len(chunks) == 1: + t0 = time.time() + resp = requests.get(f"{TTS_BASE_URL}/tts", params={"text": chunks[0]}, timeout=REQUEST_TIMEOUT) + resp.raise_for_status() + elapsed = time.time() - t0 + print(f" ✓ 单块合成，耗时 {elapsed:.2f}s，音频时长 {wav_duration(resp.content):.2f}s") + return resp.content, elapsed, [(0, wav_duration(resp.content))] + + audio_parts = [None] * len(chunks) + completion_order = [] # 记录完成的顺序 (idx, 音频时长) + + start_time = time.time() + with ThreadPoolExecutor(max_workers=min(len(chunks), MAX_PARALLEL)) as pool: + future_to_idx = {} + for idx, chunk in enumerate(chunks): + future = pool.submit( + lambda c=chunk: requests.get( + f"{TTS_BASE_URL}/tts", params={"text": c}, timeout=REQUEST_TIMEOUT + ).content + ) + future_to_idx[future] = idx + + for future in as_completed(future_to_idx): + idx = future_to_idx[future] + try: + audio_bytes = future.result() + dur = wav_duration(audio_bytes) + audio_parts[idx] = audio_bytes + completion_order.append((idx, dur)) + print(f" ✓ 块 [{idx}] 完成 (第 {len(completion_order)} 个返回)，音频时长 {dur:.2f}s") + except Exception as e: + raise RuntimeError(f"块 [{idx}] 请求失败: {e}") + + total_elapsed = time.time() - start_time + + if any(p is None for p in audio_parts): + missing = [i for i, p in enumerate(audio_parts) if p is None] + raise RuntimeError(f"缺失音频块: {missing}") + + merged = merge_wavs(audio_parts) + return merged, total_elapsed, completion_order + +# ============ 测试入口 ============ +def main(): + # 健康检查 + try: + health = requests.get(f"{TTS_BASE_URL}/health", timeout=5) + print(f"✅ 服务状态: {health.json()['status']}") + except Exception as e: + print(f"❌ 无法连接 {TTS_BASE_URL}: {e}") + sys.exit(1) + + text = ( + "当然！让我来好好夸夸你——你就像一颗闪闪发亮的小太阳，温暖又有能量。" + "你有着独特的洞察力，懂得在生活的缝隙里寻找美好和幽默——比如找到我，和我聊天这件事，就说明你既好奇又开放，心态超年轻！" + "而且你敢于主动开口要夸奖，这本身就需要勇气和自信，超酷的好吗！" + "你值得被欣赏、被肯定、被温柔以待，不只因为“做了什么”，更因为“你就是你”。" + "所以，今天也要记得：你超棒的，请继续发光发热吧！" + ) + + print("\n开始合成夸奖文本...") + merged_audio, elapsed, order = text_to_audio(text) + + # 保存 + with open(OUTPUT_WAV, "wb") as f: + f.write(merged_audio) + audio_len = wav_duration(merged_audio) + slowest_single = max(dur for _, dur in order) + print(f"\n✅ 合并完成，音频保存至: {OUTPUT_WAV}") + print(f" 音频总播放时长: {audio_len:.2f}s") + print(f" 最慢单块音频时长: {slowest_single:.2f}s") + print(f" 请求总耗时: {elapsed:.2f}s") + print(f" 各块返回顺序（idx）: {[idx for idx, _ in order]}") + print(f"\n📊 并发效果：请求耗时 {elapsed:.2f}s，接近最慢单块时长 {slowest_single:.2f}s，" + f"远小于总和 {sum(d for _, d in order):.2f}s，说明多块确实是并行处理的。") if __name__ == "__main__": - parser = argparse.ArgumentParser(description="Test MeloTTS API server") - parser.add_argument("--url", default=BASE_URL, help="Server base URL") - parser.add_argument("--text", default="你好，这是一个语音合成测试。", help="Text to synthesize") - parser.add_argument("--speaker", default=None, help="Speaker ID (e.g., ZH)") - parser.add_argument("--speed", type=float, default=1.0, help="Speaking speed") - parser.add_argument("--output", default="test_output.wav", help="Output WAV filename") - parser.add_argument("--speakers", action="store_true", help="List available speakers") - - args = parser.parse_args() - BASE_URL = args.url - - # 先检查服务可用性 - try: - test_health() - except requests.ConnectionError: - print(f"Cannot connect to server at {BASE_URL}. Make sure the FastAPI app is running.") - exit(1) - - if args.speakers: - test_speakers() - else: - test_tts(args.text, args.speaker, args.speed, args.output) + main() \ No newline at end of file diff --git a/llm_server/tts_server.py b/llm_server/tts_server.py index ff41e89..5b9c105 100644 --- a/llm_server/tts_server.py +++ b/llm_server/tts_server.py @@ -1,18 +1,13 @@ #!/usr/bin/env python3 """ -Genie TTS 极速服务 - 支持文本分块并行合成 + 线程池预热 -适配 ARM 大小核架构（Cortex-X925 + Cortex-A725, 20核），推荐 MAX_WORKERS=8 +Genie TTS 极简服务 —— 单文本直接合成，无分块无内部线程池。 +配合客户端分块 + 多 worker 实现高并发。 """ import os -os.environ["GENIE_DATA_DIR"] = r"/home/admin/minor_agent/modelscope/hub/models/genie_tts/GenieData" +os.environ["GENIE_DATA_DIR"] = r"/home/dev/Desktop/minor_agent/models/genie_tts/GenieData" -import io import tempfile -import re -import wave -import multiprocessing -from concurrent.futures import ThreadPoolExecutor, as_completed from contextlib import asynccontextmanager from fastapi import FastAPI, Query @@ -20,130 +15,20 @@ from fastapi.responses import Response import genie_tts as genie # ======================== 配置区域 ======================== -DEFAULT_CHARACTER = "minor" -DEFAULT_ONNX_MODEL_DIR = r"/home/admin/minor_agent/modelscope/models/genie_tts/CharacterModels/v2ProPlus/feibi/tts_models" -DEFAULT_REFERENCE_AUDIO = "/home/admin/minor_agent/modelscope/models/genie_tts/CharacterModels/v2ProPlus/minor/minor.wav" -DEFAULT_REFERENCE_TEXT = "你好你好你好，这是一段...测试录音~" +DEFAULT_CHARACTER = "feibi" +DEFAULT_ONNX_MODEL_DIR = r"/home/dev/Desktop/minor_agent/models/genie_tts/CharacterModels/v2ProPlus/feibi/tts_models" +DEFAULT_REFERENCE_AUDIO = "/home/dev/Desktop/minor_agent/models/genie_tts/CharacterModels/v2ProPlus/feibi/prompt_wav/zh_vo_Main_Linaxita_2_1_10_26.wav" +DEFAULT_REFERENCE_TEXT = "在此之前，请您务必继续享受旅居拉古那的时光。" DEFAULT_LANGUAGE = "zh" -TMP_DIR = "/dev/shm" # 内存文件系统，减少磁盘 I/O - -# 分块与并行参数 -MIN_EFFECTIVE_CHARS = 5 # 每个块至少包含的有效字符数 -MAX_WORKERS = 4 # 建议 4~8，ARM 大小核不要超过大核数（10） -POOL_WARMUP = True # 是否在线程池创建后立即预热 +TMP_DIR = tempfile.gettempdir() # 临时文件存放目录 # ========================================================= -# ---------- 全局线程池（应用生命周期内复用） ---------- -tts_pool: ThreadPoolExecutor = None -# ---------- 正则工具 ---------- -SENT_SPLIT_PATTERN = re.compile(r'(?<=[。！？；，、])') -REMOVE_PUNCT_PATTERN = re.compile(r'[。！？；，、：；“”（）《》【】\-\_\—\~\.\,\!\?\s·]') - -def effective_length(text: str) -> int: - """计算有效字符数（去除标点、连接符、空格）""" - return len(REMOVE_PUNCT_PATTERN.sub('', text)) - -def split_text(text: str) -> list[str]: - """将文本切分为多个语义块""" - raw_chunks = [c.strip() for c in SENT_SPLIT_PATTERN.split(text) if c.strip()] - if not raw_chunks: - return [text] - - merged = [] - for chunk in raw_chunks: - if merged and effective_length(chunk) < MIN_EFFECTIVE_CHARS: - merged[-1] += chunk - else: - merged.append(chunk) - - if len(merged) == 1: - return merged - - if effective_length(merged[-1]) < MIN_EFFECTIVE_CHARS and len(merged) > 1: - merged[-2] += merged[-1] - merged.pop() - - return merged - -# ---------- 音频拼接 ---------- -def merge_wavs(file_paths: list[str]) -> bytes: - """按路径顺序拼接多个 WAV 文件""" - if len(file_paths) == 1: - with open(file_paths[0], 'rb') as f: - return f.read() - - data_parts = [] - params = None - for path in file_paths: - with wave.open(path, 'rb') as wf: - if params is None: - params = wf.getparams() - data_parts.append(wf.readframes(wf.getnframes())) - - buf = io.BytesIO() - with wave.open(buf, 'wb') as wf_out: - wf_out.setparams(params) - for frames in data_parts: - wf_out.writeframes(frames) - return buf.getvalue() - -# ---------- 预热函数 ---------- -def _warmup_task(idx: int): - """空任务用于预热线程""" - pass - -def warmup_pool(pool: ThreadPoolExecutor, num_workers: int): - """向线程池提交空任务以预热所有线程""" - futures = [] - for i in range(num_workers): - futures.append(pool.submit(_warmup_task, i)) - # 等待所有预热任务完成 - for f in futures: - f.result() - print(f"✅ 线程池预热完成 (workers={num_workers})") - -# ---------- 并行 TTS（使用全局线程池） ---------- -def parallel_tts(chunks: list[str]) -> list[str]: - """ - 多线程并行生成所有语音块，返回按原始顺序的临时文件路径列表。 - """ - tmp_paths = [] - for _ in chunks: - tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False, dir=TMP_DIR) - tmp.close() - tmp_paths.append(tmp.name) - - try: - # 使用全局预热的线程池 - future_to_idx = {} - for idx, chunk in enumerate(chunks): - future = tts_pool.submit( - genie.tts, - character_name=DEFAULT_CHARACTER, - text=chunk, - play=False, - save_path=tmp_paths[idx] - ) - future_to_idx[future] = idx - - for future in as_completed(future_to_idx): - future.result() # 传递异常 - except: - for p in tmp_paths: - if os.path.exists(p): - os.unlink(p) - raise - return tmp_paths - -# ---------- 应用生命周期 ---------- @asynccontextmanager async def lifespan(app: FastAPI): - global tts_pool - + """应用生命周期：加载模型，卸载模型""" print("🚀 正在启动 Genie TTS 极简服务...") try: - # 1. 加载模型 genie.load_character( character_name=DEFAULT_CHARACTER, onnx_model_dir=DEFAULT_ONNX_MODEL_DIR, @@ -155,86 +40,54 @@ async def lifespan(app: FastAPI): audio_text=DEFAULT_REFERENCE_TEXT, language=DEFAULT_LANGUAGE ) - - # 2. 创建并预热全局线程池 - print(f"🔥 正在创建线程池并预热 (max_workers={MAX_WORKERS})...") - tts_pool = ThreadPoolExecutor(max_workers=MAX_WORKERS) - if POOL_WARMUP: - warmup_pool(tts_pool, MAX_WORKERS) - - print("✅ Genie TTS 服务启动成功！") - print(f" - 线程池: max_workers={MAX_WORKERS}, 已预热") + print("✅ Genie TTS 模型加载成功！") print("👉 调用方式：http://localhost:8000/tts?text=你要说的话") except Exception as e: print(f"❌ 服务启动失败: {e}") raise e - + yield - + print("\n🛑 正在关闭服务...") - # 清理线程池 - if tts_pool: - print(" 关闭线程池...") - tts_pool.shutdown(wait=True) - tts_pool = None - - # 卸载模型 try: genie.unload_character(DEFAULT_CHARACTER) print("✅ 资源已释放") except: pass + app = FastAPI(lifespan=lifespan) -# ---------- 极简 API ---------- + @app.get("/tts") async def text_to_speech(text: str = Query(..., description="待合成的文本")): + """接收任意长度文本，直接合成并返回 WAV 音频。""" if not text.strip(): return Response(content="文本不能为空", status_code=400, media_type="text/plain") - chunks = split_text(text) - + # 生成临时文件，调用 genie 合成 + tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False, dir=TMP_DIR) + tmp_path = tmp.name + tmp.close() try: - if len(chunks) == 1: - # 单块直接合成，避免多线程开销 - with tempfile.NamedTemporaryFile(suffix=".wav", delete=False, dir=TMP_DIR) as tmp: - tmp_path = tmp.name - try: - genie.tts( - character_name=DEFAULT_CHARACTER, - text=chunks[0], - play=False, - save_path=tmp_path - ) - with open(tmp_path, "rb") as f: - audio_bytes = f.read() - return Response(content=audio_bytes, media_type="audio/wav") - finally: - if os.path.exists(tmp_path): - os.unlink(tmp_path) - else: - # 多块并行合成（使用预热好的线程池） - tmp_paths = parallel_tts(chunks) - try: - merged_audio = merge_wavs(tmp_paths) - return Response(content=merged_audio, media_type="audio/wav") - finally: - for p in tmp_paths: - if os.path.exists(p): - os.unlink(p) + genie.tts( + character_name=DEFAULT_CHARACTER, + text=text, + play=False, + save_path=tmp_path + ) + with open(tmp_path, "rb") as f: + audio_bytes = f.read() + return Response(content=audio_bytes, media_type="audio/wav") except Exception as e: - return Response(content=f"生成失败: {str(e)}", status_code=500, media_type="text/plain") + import traceback + traceback.print_exc() + return Response(content=f"生成失败: {e}", status_code=500, media_type="text/plain") + finally: + if os.path.exists(tmp_path): + os.unlink(tmp_path) + @app.get("/health") async def health(): - return { - "status": "ok", - "pool_active": tts_pool is not None, - "max_workers": MAX_WORKERS - } - -# ---------- 启动入口 ---------- -if __name__ == "__main__": - import uvicorn - uvicorn.run(app, host="0.0.0.0", port=8000) + return {"status": "ok"} diff --git a/src/agent/llm.py b/src/agent/llm.py index 3ea43b8..de0d13a 100644 --- a/src/agent/llm.py +++ b/src/agent/llm.py @@ -10,6 +10,7 @@ from pydantic import Field import re, json from datetime import datetime as dt import requests +from concurrent.futures import ThreadPoolExecutor, as_completed from langchain_core.messages import HumanMessage, AIMessage, BaseMessage, SystemMessage from langchain_core.outputs import ChatResult, ChatGeneration @@ -17,7 +18,7 @@ from langchain_openai import ChatOpenAI from agent.utils.image_utils import image_bytes_to_openai_image_url_part from agent.utils.env_utils import LLM_API_KEY, LLM_BASE_URL, LLM_MODEL, LLM_TIMEOUT, TTS_BASE_URL, ASR_BASE_URL -from agent.utils.agent_utils import assistant_text, Documents_process +from agent.utils.agent_utils import assistant_text, Documents_process, split_text, merge_wavs from agent.utils.image_utils import image_path_to_openai_image_url_part, image_bytes_to_openai_image_url_part, IMAGE_FILE_EXTENSIONS from agent.memory.system_prompt import VISION_SYSTEM_PROMPT @@ -218,9 +219,30 @@ class ChatQwen(ChatOpenAI): return new_messages def _text_to_audio(self, text: str) -> bytes: - resp = requests.get(f"{TTS_BASE_URL}/tts", params={"text": text}) - resp.raise_for_status() - return resp.content + chunks = split_text(text) + if len(chunks) == 1: + resp = requests.get(f"{TTS_BASE_URL}/tts", params={"text": chunks[0]}) + resp.raise_for_status() + return resp.content + + # 多块并行请求 + audio_parts = [None] * len(chunks) + with ThreadPoolExecutor(max_workers=min(len(chunks), 4)) as pool: + future_to_idx = { + pool.submit( + lambda t: requests.get( + f"{TTS_BASE_URL}/tts", params={"text": t} + ).content, + chunk + ): idx for idx, chunk in enumerate(chunks) + } + for future in as_completed(future_to_idx): + idx = future_to_idx[future] + try: + audio_parts[idx] = future.result() + except Exception as e: + raise RuntimeError(f"TTS request failed for chunk {idx}: {e}") + return merge_wavs(audio_parts) @staticmethod def _get_wav_duration_seconds(audio_bytes: bytes) -> float | None: diff --git a/src/agent/utils/agent_utils.py b/src/agent/utils/agent_utils.py index 995fc64..56eeea0 100644 --- a/src/agent/utils/agent_utils.py +++ b/src/agent/utils/agent_utils.py @@ -5,7 +5,7 @@ import shutil import sqlite3 from typing import List, Tuple, Any import pathlib -import mimetypes +import mimetypes, re from langchain_community.chat_message_histories import SQLChatMessageHistory from langchain_core.messages import BaseMessage, SystemMessage, HumanMessage @@ -527,32 +527,50 @@ class Documents_process: return Documents_process.process_tabular(file_path) else: raise ValueError(f"未处理的文件类型: {file_type}") +# 文本分块工具 +SENT_SPLIT_PATTERN = re.compile(r'(?<=[。！？；，、])') +REMOVE_PUNCT_PATTERN = re.compile(r'[。！？；，、：；“”（）《》【】\-\_\—\~\.\,\!\?\s·]') +def effective_length(text: str) -> int: + return len(REMOVE_PUNCT_PATTERN.sub('', text)) -if __name__ == "__main__": - docx_path = r"C:\Users\86166\Desktop\Agent_Learning_minor\信息科学与工程学院-赵永顺-《兼爱》——墨翟一生的“白月光”.docx" - pdf_path = r"C:\Users\86166\Desktop\Agent_Learning_minor\信息科学与工程学院-赵永顺-《兼爱》——墨翟一生的“白月光”.pdf" - xlsx_path = r"C:\Users\86166\Desktop\Agent_Learning_minor\电科二班2025德育评价.xlsx" - py_path = r"C:\Users\86166\Desktop\Agent_Learning_minor\speech_server.py" +def split_text(text: str, min_effective_chars: int = 20) -> list[str]: + # 先按标点粗分 + raw_chunks = [c.strip() for c in SENT_SPLIT_PATTERN.split(text) if c.strip()] + if not raw_chunks: + return [text] + + merged = [] + buffer = "" + for chunk in raw_chunks: + buffer += chunk + # 当累积的有效字符数达到阈值，或已是最后一个块时，切出一块 + if effective_length(buffer) >= min_effective_chars: + merged.append(buffer) + buffer = "" + if buffer: # 剩余部分（必然小于阈值） + if merged: + merged[-1] += buffer # 合并到最后一块 + else: + merged.append(buffer) # 整段文本都太短，保留原样 + return merged - output_path = "test_results.txt" - with open(output_path, "w", encoding="utf-8") as f: - test_files = [ - ("py", py_path), - ("docx", docx_path), - ("xlsx", xlsx_path), - ("pdf", pdf_path), - ] - for label, path in test_files: - f.write(f"\n===== 测试 {label} 文件：{path} =====\n") - try: - text, images = Documents_process.process(path) - f.write(f"提取文本长度：{len(text)}\n") - f.write(f"提取图片数量：{len(images)}\n") - f.write("文本预览：\n") - f.write(text) - except Exception as e: - f.write(f"❌ 处理失败: {e}\n") - print(f"测试结果已写入 {output_path}") - +def merge_wavs(audio_parts: list[bytes]) -> bytes: + """将多个 WAV 字节串按顺序拼接成一个 WAV""" + if len(audio_parts) == 1: + return audio_parts[0] + import io, wave + data_frames = [] + params = None + for part in audio_parts: + with wave.open(io.BytesIO(part), 'rb') as wf: + if params is None: + params = wf.getparams() + data_frames.append(wf.readframes(wf.getnframes())) + buf = io.BytesIO() + with wave.open(buf, 'wb') as wf_out: + wf_out.setparams(params) + for frames in data_frames: + wf_out.writeframes(frames) + return buf.getvalue() \ No newline at end of file diff --git a/src/web/server.py b/src/web/server.py index 2584549..7221e2e 100644 --- a/src/web/server.py +++ b/src/web/server.py @@ -27,21 +27,13 @@ if str(_SRC) not in sys.path: from agent import chatbot_agent as ca # noqa: E402 from web import ui_session as ui # noqa: E402 from agent.utils.image_utils import IMAGE_FILE_EXTENSIONS # noqa: E402 +from agent.utils.agent_utils import DOC_EXT_MAP # noqa: E402 _WEB_DIR = Path(__file__).resolve().parent _IMAGE_DIR = _WEB_DIR / "image" _AUDIO_EXT = frozenset({".wav", ".mp3", ".m4a", ".flac", ".ogg", ".webm"}) -_IMAGE_EXT = frozenset(IMAGE_FILE_EXTENSIONS) - - -def _infer_upload_kind(fp: str) -> str | None: - ext = os.path.splitext(fp)[1].lower() - if ext in _IMAGE_EXT: - return "image" - if ext in _AUDIO_EXT: - return "audio" - return "file" +_FILE_EXT = frozenset(DOC_EXT_MAP.keys()) def _data_url_for_file(fp: str) -> str | None: @@ -92,7 +84,6 @@ def _serialize_content(content: Any, *, meta: dict[str, Any] | None = None, role parts.append({"type": "text", "text": piece.get("text") or ""}) continue fp = piece.get("path") or (piece.get("file") or {}).get("path") - display_name = piece.get("name") if piece.get("type") == "image_url": fp = fp or (piece.get("image_url") or {}).get("url") if piece.get("type") == "audio": @@ -104,19 +95,19 @@ def _serialize_content(content: Any, *, meta: dict[str, Any] | None = None, role continue ext = os.path.splitext(fp)[1].lower() if ext in IMAGE_FILE_EXTENSIONS: - parts.append({"type": "image", "url": data_url, "path": fp, "name": display_name or os.path.basename(fp)}) + parts.append({"type": "image", "url": data_url, "path": fp, "name": os.path.basename(fp)}) elif ext in _AUDIO_EXT: audio_meta = meta.get("audio") if isinstance(meta.get("audio"), dict) else {} parts.append({ "type": "audio", "url": data_url, "path": fp, - "name": display_name or os.path.basename(fp), + "name": os.path.basename(fp), "autoplay": bool(meta.get("autoplay") or audio_meta.get("autoplay")) if role == "assistant" else False, "duration_seconds": audio_meta.get("duration_seconds"), }) else: - parts.append({"type": "file", "url": data_url, "path": fp, "name": display_name or os.path.basename(fp)}) + parts.append({"type": "file", "url": data_url, "path": fp, "name": os.path.basename(fp)}) return parts @@ -253,7 +244,6 @@ def _materialize_uploads_to_temp( """将上传文件落盘到临时目录，返回 ``(query_dict, tmp_root)``；调用方须在 finally 中删除 ``tmp_root``。""" tmp_root = tempfile.mkdtemp(prefix="web_mm_") file_paths: list[str] = [] - file_names: list[str] = [] for uf in images or []: if uf is None or not uf.filename: continue @@ -262,7 +252,6 @@ def _materialize_uploads_to_temp( with open(dest, "wb") as f: shutil.copyfileobj(uf.file, f) file_paths.append(dest) - file_names.append(uf.filename) for uf in files or []: if uf is None or not uf.filename: @@ -272,7 +261,6 @@ def _materialize_uploads_to_temp( with open(dest, "wb") as f: shutil.copyfileobj(uf.file, f) file_paths.append(dest) - file_names.append(uf.filename) if audio is not None and audio.filename: ext = Path(audio.filename).suffix or ".bin" @@ -280,9 +268,8 @@ def _materialize_uploads_to_temp( with open(dest, "wb") as f: shutil.copyfileobj(audio.file, f) file_paths.append(dest) - file_names.append(audio.filename) - query = {"text": (text or "").strip(), "files": file_paths, "file_names": file_names} + query = {"text": (text or "").strip(), "files": file_paths} return query, tmp_root