From 87ee138fb5b9f1e7a1bc9d4eadae06ec8cf066b2 Mon Sep 17 00:00:00 2001 From: minorQAQ <202200120184@mail.sdu.edu.cn> Date: Sat, 16 May 2026 18:55:44 +0800 Subject: [PATCH] =?UTF-8?q?rag=E5=9F=BA=E5=BA=A7=E5=B7=B2=E7=BB=8F?= =?UTF-8?q?=E6=90=AD=E5=BB=BA=E5=A5=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env | 3 +- README.md | 46 ++++--- description.txt | 1 - llm_server/rag_server.py | 165 +++++++++++++++++++++++ llm_server/speech_server.py | 1 + llm_server/start_asr.sh | 86 +++++------- llm_server/start_rag_server.sh | 37 +++++ llm_server/start_tts.sh | 35 +++++ llm_server/stop_asr.sh | 14 -- llm_server/stop_audio.sh | 14 -- llm_server/stop_tts.sh | 0 llm_server/test.py | 148 -------------------- llm_server/test_asr.py | 115 ++++++++++++++++ llm_server/test_rag.py | 82 +++++++++++ llm_server/test_tts.py | 71 ++++++++++ llm_server/tts_server.py | 240 +++++++++++++++++++++++++++++++++ src/agent/utils/env_utils.py | 9 +- 17 files changed, 816 insertions(+), 251 deletions(-) delete mode 100644 description.txt create mode 100644 llm_server/rag_server.py create mode 100644 llm_server/start_rag_server.sh delete mode 100644 llm_server/stop_asr.sh delete mode 100644 llm_server/stop_audio.sh delete mode 100644 llm_server/stop_tts.sh delete mode 100644 llm_server/test.py create mode 100644 llm_server/test_asr.py create mode 100644 llm_server/test_rag.py create mode 100644 llm_server/test_tts.py create mode 100644 llm_server/tts_server.py diff --git a/.env b/.env index 04a82ab..01c51a8 100644 --- a/.env +++ b/.env @@ -9,8 +9,9 @@ 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" -TTS_BASE_URL = "http://192.168.137.96:8902" 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" #6000pro： LLM_API_KEY='semi-ai.tech' diff --git a/README.md b/README.md index d691334..2aa83c7 100644 --- a/README.md +++ b/README.md @@ -1,30 +1,42 @@ -cd .\Agent\ +# 进入项目目录并提交代码（Windows 下路径为 .\Agent\） +cd Agent git add . git commit -m "修改内容的说明" git push +# 创建并激活虚拟环境（Linux/macOS 用 source，Windows 用 myenv\Scripts\activate） python -m venv myenv -source myenv/bin/activate || myenv/Scripts/activate -python -m pip install --upgrade pip -pip install -U langchain langchain-openai dotenv -pip install -e .(详情见pyproject.toml) -pip install langgraph-cli[inmem] -pip install sqlalchemy pymysql loguru psycopg2-binary(postgresql) *可选数据库开发 +source myenv/bin/activate || myenv/Scripts/activate -# Environment -Copy `.env.example` to `.env` and fill in the required values before running the app. +# 升级 pip 并安装核心依赖 +python -m pip install --upgrade pip +pip install -U langchain langchain-openai python-dotenv + +# 以可编辑模式安装当前项目（配置见 pyproject.toml） +pip install -e . + +# 安装 LangGraph CLI（开发模式） +pip install "langgraph-cli[inmem]" + +# 可选：安装数据库相关依赖（psycopg2-binary 用于 PostgreSQL） +pip install sqlalchemy pymysql loguru psycopg2-binary + +# 环境变量配置 +# Copy .env.example to .env and fill in the required values before running the app. cp .env.example .env -# Required variables -# - LLM_BASE_URL -# - LLM_API_KEY -# - LLM_MODEL +# 必须填写的变量： +# .env.example -# Web 对话（推荐）：在 ``Agent`` 目录下保证 ``PYTHONPATH`` 包含 ``src`` 后执行 +# 启动 Web 对话服务（推荐） +# 确保在 Agent 目录下，且 PYTHONPATH 包含 src 目录 cd ./Agent +export PYTHONPATH=src:$PYTHONPATH # Linux/macOS +# 或 Windows PowerShell: $env:PYTHONPATH="src;$env:PYTHONPATH" uvicorn agent.web.server:app --host 127.0.0.1 --port 8765 -# 发送消息为两阶段 API：``POST /api/chat/start``（立即落盘用户消息）+ ``POST /api/chat/complete``（推理）；前端已按此顺序调用。 -cloudflared tunnel --url http://localhost:8765开公网 - +# API 为两阶段调用：POST /api/chat/start（落盘用户消息） + POST /api/chat/complete（推理） +# 前端已按此顺序调用。 +# （可选）通过 Cloudflare Tunnel 暴露公网 +cloudflared tunnel --url http://localhost:8765 \ No newline at end of file diff --git a/description.txt b/description.txt deleted file mode 100644 index 729d1a2..0000000 --- a/description.txt +++ /dev/null @@ -1 +0,0 @@ -目前版本已实现多模态聊天加tool calling \ No newline at end of file diff --git a/llm_server/rag_server.py b/llm_server/rag_server.py new file mode 100644 index 0000000..feec6dc --- /dev/null +++ b/llm_server/rag_server.py @@ -0,0 +1,165 @@ +import torch +import torch.nn.functional as F +from fastapi import FastAPI +from pydantic import BaseModel +from typing import List +from modelscope import AutoTokenizer, AutoModel, AutoModelForCausalLM +from transformers import BitsAndBytesConfig + +# 在 load_models() 里添加量化配置 +quantization_config = BitsAndBytesConfig(load_in_8bit=True) +# ===================== 模型路径 ===================== +EMBED_MODEL_PATH = "/home/admin/minor_agent/modelscope/models/Qwen/Qwen3-Embedding-0___6B" +RERANK_MODEL_PATH = "/home/admin/minor_agent/modelscope/models/Qwen/Qwen3-Reranker-4B" + +# ===================== 初始化 FastAPI ===================== +app = FastAPI(title="RAG Backend with 8-bit Qwen3 Models") + +# ===================== 全局变量（模型、分词器） ===================== +embed_tokenizer = None +embed_model = None +rerank_tokenizer = None +rerank_model = None + +# ===================== 工具函数 ===================== +def last_token_pool(last_hidden_states, attention_mask): + left_padding = (attention_mask[:, -1].sum() == attention_mask.shape[0]) + if left_padding: + return last_hidden_states[:, -1] + else: + sequence_lengths = attention_mask.sum(dim=1) - 1 + batch_size = last_hidden_states.shape[0] + return last_hidden_states[torch.arange(batch_size, device=last_hidden_states.device), sequence_lengths] + +def get_detailed_instruct(task_description: str, query: str) -> str: + return f'Instruct: {task_description}\nQuery:{query}' + +def format_rerank_instruction(instruction, query, doc): + if instruction is None: + instruction = 'Given a web search query, retrieve relevant passages that answer the query' + return f": {instruction}\n: {query}\n: {doc}" + +# ===================== 启动时加载模型（8‑bit） ===================== +@app.on_event("startup") +def load_models(): + global embed_tokenizer, embed_model, rerank_tokenizer, rerank_model + + # ---------- 加载 Embedding 模型 ---------- + print("Loading Embedding model with 16-bit quantization...") + embed_tokenizer = AutoTokenizer.from_pretrained(EMBED_MODEL_PATH, padding_side='left') + embed_model = AutoModel.from_pretrained( + EMBED_MODEL_PATH, + #load_in_8bit=True, # 启用 8-bit 量化 + device_map="auto", # 自动分配到 GPU + dtype=torch.float16 # 计算精度保持 fp16 + ) + embed_model.eval() + print("Embedding model loaded.") + + # ---------- 加载 Reranker 模型 ---------- + print("Loading Reranker model with 8-bit quantization...") + rerank_tokenizer = AutoTokenizer.from_pretrained(RERANK_MODEL_PATH, padding_side='left') + rerank_model = AutoModelForCausalLM.from_pretrained( + RERANK_MODEL_PATH, + quantization_config=quantization_config, + device_map="auto", + dtype=torch.float16 + ) + rerank_model.eval() + + # 预计算 Reranker 所需的固定 tokens + 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 + + 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" + prefix_tokens = rerank_tokenizer.encode(prefix, add_special_tokens=False) + suffix_tokens = rerank_tokenizer.encode(suffix, add_special_tokens=False) + + print("Reranker model loaded.") + +# ===================== API 请求体定义 ===================== +class EmbedRequest(BaseModel): + task_description: str = "Given a web search query, retrieve relevant passages that answer the query" + queries: List[str] + documents: List[str] # 可选，若只查询文本可只传 queries + +class EmbedResponse(BaseModel): + query_embeddings: List[List[float]] + doc_embeddings: List[List[float]] # 当传入 documents 时返回 + +class RerankRequest(BaseModel): + task_description: str = "Given a web search query, retrieve relevant passages that answer the query" + query: str + documents: List[str] + +class RerankResponse(BaseModel): + scores: List[float] + +# ===================== Embedding 端点 ===================== +@app.post("/embed", response_model=EmbedResponse) +async def get_embeddings(req: EmbedRequest): + # 为 query 加上指令，document 不加指令 + instructed_queries = [get_detailed_instruct(req.task_description, q) for q in req.queries] + # 所有文本拼成一个 batch：先是 queries，后是 documents（如果有） + input_texts = instructed_queries + req.documents + + # 分词 + batch_dict = embed_tokenizer( + input_texts, + padding=True, + truncation=True, + max_length=8192, + return_tensors="pt", + ).to(embed_model.device) + + with torch.no_grad(): + outputs = embed_model(**batch_dict) + embeddings = last_token_pool(outputs.last_hidden_state, batch_dict['attention_mask']) + embeddings = F.normalize(embeddings, p=2, dim=1) + + # 拆分 query 和 document 的嵌入 + nq = len(instructed_queries) + query_embs = embeddings[:nq].cpu().tolist() + doc_embs = embeddings[nq:].cpu().tolist() if req.documents else [] + + return EmbedResponse(query_embeddings=query_embs, doc_embeddings=doc_embs) + +# ===================== Reranker 端点 ===================== +@app.post("/rerank", response_model=RerankResponse) +async def rerank_documents(req: RerankRequest): + # 构造带格式的输入对 + pairs = [ + format_rerank_instruction(req.task_description, req.query, doc) + for doc in req.documents + ] + + # 处理输入，添加 prefix/suffix tokens 并 pad + inputs = rerank_tokenizer( + pairs, + padding=False, + truncation='longest_first', + return_attention_mask=False, + max_length=max_rerank_length - len(prefix_tokens) - len(suffix_tokens) + ) + for i, ele in enumerate(inputs['input_ids']): + inputs['input_ids'][i] = prefix_tokens + ele + suffix_tokens + inputs = rerank_tokenizer.pad(inputs, padding=True, return_tensors="pt", max_length=max_rerank_length) + inputs = {k: v.to(rerank_model.device) for k, v in inputs.items()} + + with torch.no_grad(): + batch_scores = rerank_model(**inputs).logits[:, -1, :] + true_vector = batch_scores[:, token_true_id] + false_vector = batch_scores[:, token_false_id] + scores = torch.stack([false_vector, true_vector], dim=1) + scores = torch.nn.functional.log_softmax(scores, dim=1) + scores = scores[:, 1].exp().cpu().tolist() + + return RerankResponse(scores=scores) + +# ===================== 健康检查 ===================== +@app.get("/health") +def health(): + return {"status": "ok"} diff --git a/llm_server/speech_server.py b/llm_server/speech_server.py index c06a74b..408ab89 100644 --- a/llm_server/speech_server.py +++ b/llm_server/speech_server.py @@ -85,3 +85,4 @@ async def text_to_speech(text: str = Query(...)): @app.get("/health") async def health(): return {"status": "ok", "models": list(models.keys())} + diff --git a/llm_server/start_asr.sh b/llm_server/start_asr.sh index e0e8764..81f713f 100644 --- a/llm_server/start_asr.sh +++ b/llm_server/start_asr.sh @@ -1,62 +1,44 @@ #!/usr/bin/env bash set -euo pipefail -# 配置区域（可根据实际情况修改） -MODEL_BASE_PATH="${MODEL_BASE_PATH:-/mnt/Qwen3-Omni-30B-A3B-Instruct-NVFP4}" -THINKER_PATH="${MODEL_BASE_PATH}/thinker" -CONTAINER_NAME="${CONTAINER_NAME:-vllm-thinker}" -HOST_PORT="${HOST_PORT:-8091}" -GPU_DEVICES="${GPU_DEVICES:-all}" # 例如 "0,1" 或 "all" -TENSOR_PARALLEL_SIZE="${TENSOR_PARALLEL_SIZE:-1}" -MAX_MODEL_LEN="${MAX_MODEL_LEN:-32768}" -VLLM_IMAGE="${VLLM_IMAGE:-vllm/vllm-openai:latest}" +# 自动获取脚本所在目录 +SCRIPT_DIR="$(cd -- "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +VENV_DIR="${SCRIPT_DIR}/.venv_audio" +MODEL_DIR="${SCRIPT_DIR}/modelscope/models/Qwen/Qwen3-ASR-1.7B" +HOST_PORT="${HOST_PORT:-8901}" -# 检查模型路径是否存在 -if [[ ! -d "$THINKER_PATH" ]]; then - echo "错误: 模型路径不存在: $THINKER_PATH" - echo "请设置环境变量 MODEL_BASE_PATH 指向正确的模型根目录（包含 thinker 子目录）" +# 激活虚拟环境 +source "${VENV_DIR}/bin/activate" + +# 检查端口占用 +if lsof -i "tcp:${HOST_PORT}" -t >/dev/null 2>&1; then + echo "Port ${HOST_PORT} already in use" >&2 exit 1 fi -# 检查 Docker 可用性 -if command -v docker >/dev/null 2>&1 && docker info >/dev/null 2>&1; then - DOCKER_CMD="docker" -else - DOCKER_CMD="sudo docker" -fi +echo "Starting vLLM server on port ${HOST_PORT} ..." +echo "Model path: ${MODEL_DIR}" -# 停止并移除可能已存在的同名容器 -${DOCKER_CMD} rm -f "${CONTAINER_NAME}" >/dev/null 2>&1 || true +# 进入脚本目录 +cd "${SCRIPT_DIR}" -echo "启动 Thinker 服务..." -echo "模型路径: $THINKER_PATH" -echo "服务端口: $HOST_PORT" -echo "Tensor Parallel Size: $TENSOR_PARALLEL_SIZE" +# 设置环境变量 +export VLLM_USE_MODELSCOPE=true +export MODELSCOPE_DOWNLOAD_MODE=local +export MODELSCOPE_HUB_ENABLE_FILE_CHECK=false +export MODELSCOPE_CACHE="${SCRIPT_DIR}/modelscope" +export HF_HUB_OFFLINE=1 +export TRANSFORMERS_OFFLINE=1 -# 启动容器 -${DOCKER_CMD} run -d \ - --name "${CONTAINER_NAME}" \ - --gpus "${GPU_DEVICES}" \ - --ipc=host \ - --network host \ - -v "${THINKER_PATH}:/model:ro" \ - -e VLLM_USE_MODELSCOPE=false \ - "${VLLM_IMAGE}" \ - sh -c "pip install --no-cache-dir nvidia-modelopt && \ - vllm serve /model \ - --port ${HOST_PORT} \ - --quantization modelopt_fp4 \ - --dtype bfloat16 \ - --max-model-len ${MAX_MODEL_LEN} \ - --tensor-parallel-size ${TENSOR_PARALLEL_SIZE} \ - --trust-remote-code \ - --enforce-eager" - -# 等待服务启动 -sleep 3 - -# 健康检查 -echo "健康检查 (等待几秒后):" -echo " curl http://127.0.0.1:${HOST_PORT}/health" -echo "查看日志:" -echo " ${DOCKER_CMD} logs -f ${CONTAINER_NAME}" \ No newline at end of file +# 启动 vLLM 服务 +vllm serve "${MODEL_DIR}" \ + --port "${HOST_PORT}" \ + --host 0.0.0.0 \ + --quantization fp8 \ + --dtype auto \ + --kv-cache-dtype fp8 \ + --max-model-len 16384 \ + --allowed-local-media-path / \ + --tensor-parallel-size 1 \ + --gpu-memory-utilization 0.10 \ + --max-num-batched-tokens 4096 diff --git a/llm_server/start_rag_server.sh b/llm_server/start_rag_server.sh new file mode 100644 index 0000000..8262baf --- /dev/null +++ b/llm_server/start_rag_server.sh @@ -0,0 +1,37 @@ +#!/usr/bin/env bash +set -euo pipefail + +# 自动获取脚本所在目录 +SCRIPT_DIR="$(cd -- "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +VENV_DIR="${VENV_DIR:-${SCRIPT_DIR}/.venv_audio}" +HOST_PORT="${HOST_PORT:-8903}" + +# 激活虚拟环境 +source "${VENV_DIR}/bin/activate" + +# 检查端口占用 +if lsof -i "tcp:${HOST_PORT}" -t >/dev/null 2>&1; then + echo "Port ${HOST_PORT} already in use" >&2 + exit 1 +fi + +echo "Starting Qwen3 RAG server on port ${HOST_PORT} ..." + +# 进入脚本目录 +cd "${SCRIPT_DIR}" + +# 离线模式设置（确保已下载模型，不从远程拉取） +export MODELSCOPE_DOWNLOAD_MODE=local +export MODELSCOPE_HUB_ENABLE_FILE_CHECK=false +export HF_HUB_OFFLINE=1 +export TRANSFORMERS_OFFLINE=1 + +# 指定 modelscope 的缓存目录（如果你的模型保存在此路径下） +# 根据你的下载路径调整，或将其设置为环境变量覆盖 +export MODELSCOPE_CACHE="${MODELSCOPE_CACHE:-/home/admin/minor_agent/modelscope}" + +# 启动 FastAPI 服务 +uvicorn rag_server:app \ + --host 0.0.0.0 \ + --port "${HOST_PORT}" \ + --workers 1 diff --git a/llm_server/start_tts.sh b/llm_server/start_tts.sh index e69de29..49ca8c1 100644 --- a/llm_server/start_tts.sh +++ b/llm_server/start_tts.sh @@ -0,0 +1,35 @@ +#!/usr/bin/env bash +set -euo pipefail + +# 自动获取脚本所在目录 +SCRIPT_DIR="$(cd -- "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +VENV_DIR="${SCRIPT_DIR}/.venv_audio" +HOST_PORT="${HOST_PORT:-8902}" + +# 激活虚拟环境 +source "${VENV_DIR}/bin/activate" + +# 检查端口占用 +if lsof -i "tcp:${HOST_PORT}" -t >/dev/null 2>&1; then + echo "Port ${HOST_PORT} already in use" >&2 + exit 1 +fi + +echo "Starting Genie TTS server on port ${HOST_PORT} ..." + +# 进入脚本目录 +cd "${SCRIPT_DIR}" + +# 设置环境变量（和 vLLM 保持一致） +export MODELSCOPE_DOWNLOAD_MODE=local +export MODELSCOPE_HUB_ENABLE_FILE_CHECK=false +export MODELSCOPE_CACHE="${SCRIPT_DIR}/modelscope" +export HF_HUB_OFFLINE=1 +export TRANSFORMERS_OFFLINE=1 + +# 启动 FastAPI 服务 +uvicorn tts_server:app \ + --host 0.0.0.0 \ + --port "${HOST_PORT}" \ + --workers 1 \ + diff --git a/llm_server/stop_asr.sh b/llm_server/stop_asr.sh deleted file mode 100644 index 3bae62e..0000000 --- a/llm_server/stop_asr.sh +++ /dev/null @@ -1,14 +0,0 @@ -#!/usr/bin/env bash -set -euo pipefail - -CONTAINER_NAME="${CONTAINER_NAME:-vllm-thinker}" - -if command -v docker >/dev/null 2>&1 && docker info >/dev/null 2>&1; then - DOCKER_CMD="docker" -else - DOCKER_CMD="sudo docker" -fi - -echo "停止并移除容器: $CONTAINER_NAME" -${DOCKER_CMD} rm -f "${CONTAINER_NAME}" >/dev/null 2>&1 || true -echo "已停止" \ No newline at end of file diff --git a/llm_server/stop_audio.sh b/llm_server/stop_audio.sh deleted file mode 100644 index b074b85..0000000 --- a/llm_server/stop_audio.sh +++ /dev/null @@ -1,14 +0,0 @@ -#!usrbinenv bash -set -euo pipefail - -HOST_PORT=${HOST_PORT-8100} - -PID=$(lsof -i tcp${HOST_PORT} -t 2devnull true) - -if [[ -z ${PID} ]]; then - echo No process found on port ${HOST_PORT} - exit 0 -fi - -echo Stopping audio server (PID ${PID}) ... -kill ${PID} \ No newline at end of file diff --git a/llm_server/stop_tts.sh b/llm_server/stop_tts.sh deleted file mode 100644 index e69de29..0000000 diff --git a/llm_server/test.py b/llm_server/test.py deleted file mode 100644 index 501aab2..0000000 --- a/llm_server/test.py +++ /dev/null @@ -1,148 +0,0 @@ -#!/usr/bin/env python3 -""" -语音服务功能测试：ASR 识别 + TTS 合成 -""" -import requests -import json -import os -import sys -import time - -HOST = "http://127.0.0.1:8100" - -def test_health(): - """1. 健康检查""" - print("=" * 50) - print("📋 健康检查") - print("=" * 50) - resp = requests.get(f"{HOST}/health") - print(f"状态码: {resp.status_code}") - print(f"模型列表: {resp.json()}") - assert resp.status_code == 200 - assert "asr" in resp.json()["models"] - assert "tts" in resp.json()["models"] - print("✅ 健康检查通过\n") - -def test_asr(audio_path=None): - """2. 测试语音识别""" - print("=" * 50) - print("🎤 语音识别测试") - print("=" * 50) - - if audio_path is None: - # 如果没有本地音频，下载一个公开示例 - import tempfile - audio_path = tempfile.mktemp(suffix=".wav") - url = "https://qianwen-res.oss-cn-beijing.aliyuncs.com/Qwen3-ASR-Repo/asr_zh.wav" - print(f"下载测试音频: {url}") - import subprocess - subprocess.run(["wget", "-q", url, "-O", audio_path], check=True) - - print(f"音频文件: {audio_path}") - with open(audio_path, "rb") as f: - resp = requests.post(f"{HOST}/asr", files={"file": f}) - - print(f"状态码: {resp.status_code}") - result = resp.json() - print(f"识别结果: {result.get('text', 'N/A')}") - assert resp.status_code == 200 - assert len(result.get("text", "")) > 0 - print("✅ 语音识别通过\n") - return result["text"] - -def test_tts(text="(激动地说)你好世界，这是一段测试语音", output="test_output.wav"): - """3. 测试语音合成""" - print("=" * 50) - print("🔊 语音合成测试") - print("=" * 50) - print(f"合成文本: {text}") - - start_time = time.time() - resp = requests.get(f"{HOST}/tts", params={"text": text}) - elapsed = time.time() - start_time - - print(f"状态码: {resp.status_code}") - print(f"耗时: {elapsed:.2f}秒") - - if resp.status_code == 200: - with open(output, "wb") as f: - f.write(resp.content) - file_size = os.path.getsize(output) / 1024 # KB - print(f"音频已保存: {output} ({file_size:.1f} KB)") - else: - print(f"错误: {resp.text}") - - assert resp.status_code == 200 - assert os.path.exists(output) - print("✅ 语音合成通过\n") - -def test_asr_english(): - """4. 测试英文识别""" - print("=" * 50) - print("🌍 英文识别测试") - print("=" * 50) - - import tempfile - audio_path = tempfile.mktemp(suffix=".wav") - url = "https://qianwen-res.oss-cn-beijing.aliyuncs.com/Qwen3-ASR-Repo/asr_en.wav" - print(f"下载英文测试音频: {url}") - import subprocess - subprocess.run(["wget", "-q", url, "-O", audio_path], check=True) - - with open(audio_path, "rb") as f: - resp = requests.post(f"{HOST}/asr", files={"file": f}) - - print(f"状态码: {resp.status_code}") - result = resp.json() - print(f"识别结果: {result.get('text', 'N/A')}") - assert resp.status_code == 200 - assert len(result.get("text", "")) > 0 - print("✅ 英文识别通过\n") - -def test_tts_long_text(): - """5. 测试长文本合成""" - print("=" * 50) - print("📜 长文本合成测试") - print("=" * 50) - - long_text = """人工智能是计算机科学的一个分支，它企图了解智能的实质， - 并生产出一种新的能以人类智能相似的方式做出反应的智能机器， - 该领域的研究包括机器人、语言识别、图像识别、自然语言处理和专家系统等。""" - - print(f"文本长度: {len(long_text)} 字") - start_time = time.time() - resp = requests.get(f"{HOST}/tts", params={"text": long_text}) - elapsed = time.time() - start_time - - print(f"状态码: {resp.status_code}") - print(f"耗时: {elapsed:.2f}秒") - - if resp.status_code == 200: - output = "test_long_output.wav" - with open(output, "wb") as f: - f.write(resp.content) - file_size = os.path.getsize(output) / 1024 - print(f"音频已保存: {output} ({file_size:.1f} KB)") - - assert resp.status_code == 200 - print("✅ 长文本合成通过\n") - -if __name__ == "__main__": - print("\n" + "🚀 开始测试语音服务".center(50, "=")) - print(f"服务地址: {HOST}\n") - - try: - test_health() - - # 可选：用自己的音频文件测试 - # test_asr("/path/to/your/audio.wav") - test_asr() - test_asr_english() - - test_tts() - test_tts_long_text() - - print("\n" + "🎉 所有测试通过！".center(50, "=")) - except Exception as e: - print(f"\n❌ 测试失败: {e}") - sys.exit(1) diff --git a/llm_server/test_asr.py b/llm_server/test_asr.py new file mode 100644 index 0000000..82da5f1 --- /dev/null +++ b/llm_server/test_asr.py @@ -0,0 +1,115 @@ +""" +Test script for Qwen3-Omni-30B-A3B-Captioner via vLLM Chat Completions API. + +Note: + This model is an audio captioner, not a pure ASR. It generates + detailed descriptions of audio content (speech, environment, music, etc.) + without any text prompt. Only the audio file is sent. + It is recommended to keep audio length <= 30 seconds for best detail. +""" + +import requests +import argparse +import os +import sys + +DEFAULT_URL = "http://127.0.0.1:8901/v1/chat/completions" + +# ---------------------------------------------------------------------- +# Optional: check audio duration (requires soundfile) +# ---------------------------------------------------------------------- +def get_audio_duration_sec(file_path: str) -> float: + """Return audio duration in seconds, or -1.0 if unable to read.""" + try: + import soundfile as sf + info = sf.info(file_path) + return info.duration + except Exception: + try: + # Fallback using librosa (if available) + import librosa + return librosa.get_duration(path=file_path) + except Exception: + return -1.0 + +# ---------------------------------------------------------------------- +def file_to_url(file_path: str) -> str: + """Convert local file path to a file:// URL accepted by vLLM.""" + abs_path = os.path.abspath(file_path) + return "file://" + abs_path.replace("\\", "/") + +def caption_audio(audio_path: str, server_url: str = DEFAULT_URL) -> str: + """ + Send audio to the Qwen3-Omni-30B-A3B-Captioner and return its description. + """ + if not os.path.exists(audio_path): + raise FileNotFoundError(f"Audio file not found: {audio_path}") + + # Check duration (warning only, not blocking) + duration = get_audio_duration_sec(audio_path) + if duration > 30: + print( + f"⚠️ Audio length is {duration:.1f}s (>30s). " + "The model may lose fine detail. Consider trimming to ≤30s.", + file=sys.stderr + ) + + audio_url = file_to_url(audio_path) + + # The model expects NO text prompt. Only the audio input is required. + payload = { + "messages": [ + { + "role": "user", + "content": [ + {"type": "audio_url", "audio_url": {"url": audio_url}} + ] + } + ] + } + + headers = {"Content-Type": "application/json"} + + try: + resp = requests.post(server_url, headers=headers, json=payload, timeout=120) + resp.raise_for_status() + result = resp.json() + # Extract assistant reply + text = result["choices"][0]["message"]["content"] + return text.strip() + except requests.exceptions.RequestException as e: + print(f"Request failed: {e}", file=sys.stderr) + if hasattr(e, 'response') and e.response is not None: + print(f"Server response: {e.response.text}", file=sys.stderr) + raise + +def main(): + parser = argparse.ArgumentParser( + description="Test Qwen3-Omni-Captioner (audio captioning, NOT pure ASR)" + ) + parser.add_argument("--audio", "-a", required=True, + help="Path to input audio file (mp3, wav, etc.)") + parser.add_argument("--url", "-u", default=DEFAULT_URL, + help="vLLM chat completions endpoint") + parser.add_argument("--output", "-o", default=None, + help="Output text file (if not provided, print to stdout)") + parser.add_argument("--verbose", "-v", action="store_true", + help="Print detailed request/response info") + + args = parser.parse_args() + + if args.verbose: + print(f"Audio file: {args.audio}") + print(f"Server URL: {args.url}") + + caption_text = caption_audio(args.audio, args.url) + + if args.output: + with open(args.output, "w", encoding="utf-8") as f: + f.write(caption_text) + print(f"Caption saved to: {args.output}") + else: + print(caption_text) + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/llm_server/test_rag.py b/llm_server/test_rag.py new file mode 100644 index 0000000..049aa74 --- /dev/null +++ b/llm_server/test_rag.py @@ -0,0 +1,82 @@ +import requests +import json + +# 服务地址 +BASE_URL = "http://0.0.0.0:8903" + +def print_separator(title): + print("\n" + "=" * 80) + print(f" {title}") + print("=" * 80) + +def test_health(): + print_separator("Health Check") + resp = requests.get(f"{BASE_URL}/health") + print(f"Status: {resp.status_code}") + print(f"Response: {resp.json()}") + +def test_embed(): + print_separator("Embedding Test") + payload = { + "task_description": "Given a web search query, retrieve relevant passages that answer the query", + "queries": [ + "What is the capital of France?", + "How to train a cat?" + ], + "documents": [ + "Paris is the capital of France.", + "Cats can be trained using positive reinforcement." + ] + } + print(">>> Request Body:") + print(json.dumps(payload, indent=2, ensure_ascii=False)) + + resp = requests.post(f"{BASE_URL}/embed", json=payload) + print(f"<<< Status: {resp.status_code}") + if resp.status_code == 200: + data = resp.json() + # 只打印向量长度和前面几个值，避免刷屏 + print(f"<<< Response Keys: {list(data.keys())}") + print(f"Number of query embeddings: {len(data['query_embeddings'])}") + print(f"Dimension of each query embedding: {len(data['query_embeddings'][0])}") + print(f"First query embedding (first 5 values): {data['query_embeddings'][0][:5]}") + print(f"Number of doc embeddings: {len(data['doc_embeddings'])}") + if data['doc_embeddings']: + print(f"Dimension of each doc embedding: {len(data['doc_embeddings'][0])}") + print(f"First doc embedding (first 5 values): {data['doc_embeddings'][0][:5]}") + else: + print("No document embeddings returned.") + # 如果需要完整输出，取消下面注释（会打印大量浮点数） + # print(f"<<< Full Response:\n{json.dumps(data, indent=2)}") + else: + print(f"Error: {resp.text}") + +def test_rerank(): + print_separator("Reranker Test") + payload = { + "task_description": "Given a web search query, retrieve relevant passages that answer the query", + "query": "What is the capital of France?", + "documents": [ + "Paris is the capital of France.", + "Cats are often difficult to train.", + "Lyon is a city in France." + ] + } + print(">>> Request Body:") + print(json.dumps(payload, indent=2, ensure_ascii=False)) + + resp = requests.post(f"{BASE_URL}/rerank", json=payload) + print(f"<<< Status: {resp.status_code}") + if resp.status_code == 200: + data = resp.json() + print(f"<<< Response Keys: {list(data.keys())}") + print(f"Scores: {data['scores']}") + # 打印完整响应 + # print(f"<<< Full Response:\n{json.dumps(data, indent=2)}") + else: + print(f"Error: {resp.text}") + +if __name__ == "__main__": + test_health() + test_embed() + test_rerank() diff --git a/llm_server/test_tts.py b/llm_server/test_tts.py new file mode 100644 index 0000000..38b28d2 --- /dev/null +++ b/llm_server/test_tts.py @@ -0,0 +1,71 @@ +import requests +import argparse +import os +import time + +# 默认服务地址 +BASE_URL = "http://127.0.0.1:8000" + +def test_health(): + """检查服务是否在线""" + resp = requests.get(f"{BASE_URL}/health") + print(f"Health check: {resp.status_code} {resp.json()}") + +def test_tts(text, speaker=None, speed=1.0, output="output.wav"): + """ + 调用 TTS 接口并保存音频文件 + """ + params = {"text": text, "speed": speed} + if speaker: + params["speaker"] = speaker + + print(f"Requesting TTS...") + start = time.time() + resp = requests.get(f"{BASE_URL}/tts", params=params) + elapsed = time.time() - start + + 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") + else: + print(f" {spk}: {resp.text[:50]}") + +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) diff --git a/llm_server/tts_server.py b/llm_server/tts_server.py new file mode 100644 index 0000000..ff41e89 --- /dev/null +++ b/llm_server/tts_server.py @@ -0,0 +1,240 @@ +#!/usr/bin/env python3 +""" +Genie TTS 极速服务 - 支持文本分块并行合成 + 线程池预热 +适配 ARM 大小核架构（Cortex-X925 + Cortex-A725, 20核），推荐 MAX_WORKERS=8 +""" + +import os +os.environ["GENIE_DATA_DIR"] = r"/home/admin/minor_agent/modelscope/hub/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 +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_LANGUAGE = "zh" +TMP_DIR = "/dev/shm" # 内存文件系统，减少磁盘 I/O + +# 分块与并行参数 +MIN_EFFECTIVE_CHARS = 5 # 每个块至少包含的有效字符数 +MAX_WORKERS = 4 # 建议 4~8，ARM 大小核不要超过大核数（10） +POOL_WARMUP = True # 是否在线程池创建后立即预热 +# ========================================================= + +# ---------- 全局线程池（应用生命周期内复用） ---------- +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, + language=DEFAULT_LANGUAGE + ) + genie.set_reference_audio( + character_name=DEFAULT_CHARACTER, + audio_path=DEFAULT_REFERENCE_AUDIO, + 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("👉 调用方式：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="待合成的文本")): + if not text.strip(): + return Response(content="文本不能为空", status_code=400, media_type="text/plain") + + chunks = split_text(text) + + 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) + except Exception as e: + return Response(content=f"生成失败: {str(e)}", status_code=500, media_type="text/plain") + +@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) diff --git a/src/agent/utils/env_utils.py b/src/agent/utils/env_utils.py index 3292bb2..d688f1e 100644 --- a/src/agent/utils/env_utils.py +++ b/src/agent/utils/env_utils.py @@ -21,12 +21,13 @@ LLM_BASE_URL = _get_env("LLM_BASE_URL", required=True) LLM_API_KEY = _get_env("LLM_API_KEY", required=True) LLM_MODEL = _get_env("LLM_MODEL", required=True) LLM_TIMEOUT = _get_env("LLM_TIMEOUT", default="60", cast=float) -LLM_REQUESTS_PER_SECOND = _get_env("LLM_REQUESTS_PER_SECOND", default="10", cast=float) -LLM_MAX_BUCKET_SIZE = _get_env("LLM_MAX_BUCKET_SIZE", default="10", cast=int) -LLM_CHECK_EVERY_N_SECONDS = _get_env("LLM_CHECK_EVERY_N_SECONDS", default="0.1", cast=float) -ASR_TTS_BASE_URL = _get_env("ASR_TTS_BASE_URL") +# LLM_REQUESTS_PER_SECOND = _get_env("LLM_REQUESTS_PER_SECOND", default="10", cast=float) +# LLM_MAX_BUCKET_SIZE = _get_env("LLM_MAX_BUCKET_SIZE", default="10", cast=int) +# LLM_CHECK_EVERY_N_SECONDS = _get_env("LLM_CHECK_EVERY_N_SECONDS", default="0.1", cast=float) +# ASR_TTS_BASE_URL = _get_env("ASR_TTS_BASE_URL") TTS_BASE_URL = _get_env("TTS_BASE_URL") ASR_BASE_URL = _get_env("ASR_BASE_URL") +RAG_BASE_URL = _get_env("RAG_BASE_URL") BROWSER_MAP: Dict[str, Tuple[str, str]] = { "edge": ("microsoft-edge", "Microsoft Edge"),