Hass/xiaozhi-esp32-server
1
0
Fork 0
mirror of https://github.com/xinnan-tech/xiaozhi-esp32-server.git synced 2026-07-28 01:53:53 +08:00
Code Issues Packages Projects Releases Wiki Activity

update:server连接api (#747)

Browse Source 
* update:server连接manager-api

* update:读取智能体模型配置

* update:添加默认模型的按钮

* update:优化配置读取方式

* update：server兼容manager接口改造

* update:优化私有配置加载

* update:加载私有模型配置
This commit is contained in:
hrz
2025-04-12 17:36:04 +08:00
committed by GitHub
parent c39ad97b8e
commit 5d69ba0796
57 changed files with 1618 additions and 1066 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main/xiaozhi-server/performance_tester.py
+181 -108
View File
@@ -5,9 +5,8 @@ from tabulate import tabulate
from typing import Dict, List
from core.utils.llm import create_instance as create_llm_instance
from core.utils.tts import create_instance as create_tts_instance
from core.utils.util import read_config
import statistics
from config.settings import get_config_file
from config.settings import load_config
import inspect
import os
import logging
@@ -18,17 +17,16 @@ logging.basicConfig(level=logging.WARNING)
class AsyncPerformanceTester:
def __init__(self):
self.config = read_config(get_config_file())
self.config = load_config()
self.test_sentences = self.config.get("module_test", {}).get(
"test_sentences",
["你好，请介绍一下你自己", "What's the weather like today?",
"请用100字概括量子计算的基本原理和应用前景"]
[
"你好，请介绍一下你自己",
"What's the weather like today?",
"请用100字概括量子计算的基本原理和应用前景",
],
)
self.results = {
"llm": {},
"tts": {},
"combinations": []
}
self.results = {"llm": {}, "tts": {}, "combinations": []}
async def _check_ollama_service(self, base_url: str, model_name: str) -> bool:
"""异步检查Ollama服务状态"""
@@ -46,7 +44,9 @@ class AsyncPerformanceTester:
data = await response.json()
models = data.get("models", [])
if not any(model["name"] == model_name for model in models):
print(f"🚫 Ollama模型 {model_name} 未找到，请先使用 ollama pull {model_name} 下载")
print(
f"🚫 Ollama模型 {model_name} 未找到，请先使用 ollama pull {model_name} 下载"
)
return False
else:
print(f"🚫 无法获取Ollama模型列表")
@@ -62,17 +62,16 @@ class AsyncPerformanceTester:
logging.getLogger("core.providers.tts.base").setLevel(logging.WARNING)
token_fields = ["access_token", "api_key", "token"]
if any(field in config and any(x in config[field] for x in ["你的", "placeholder"]) for field in
token_fields):
if any(
field in config
and any(x in config[field] for x in ["你的", "placeholder"])
for field in token_fields
):
print(f"⏭️ TTS {tts_name} 未配置access_token/api_key，已跳过")
return {"name": tts_name, "type": "tts", "errors": 1}
module_type = config.get('type', tts_name)
tts = create_tts_instance(
module_type,
config,
delete_audio_file=True
)
module_type = config.get("type", tts_name)
tts = create_tts_instance(module_type, config, delete_audio_file=True)
print(f"🎵 测试 TTS: {tts_name}")
@@ -103,7 +102,7 @@ class AsyncPerformanceTester:
"name": tts_name,
"type": "tts",
"avg_time": total_time / test_count,
"errors": 0
"errors": 0,
}
except Exception as e:
@@ -115,8 +114,8 @@ class AsyncPerformanceTester:
try:
# 对于Ollama，跳过api_key检查并进行特殊处理
if llm_name == "Ollama":
base_url = config.get('base_url', 'http://localhost:11434')
model_name = config.get('model_name')
base_url = config.get("base_url", "http://localhost:11434")
model_name = config.get("model_name")
if not model_name:
print(f"🚫 Ollama未配置model_name")
return {"name": llm_name, "type": "llm", "errors": 1}
@@ -124,21 +123,27 @@ class AsyncPerformanceTester:
if not await self._check_ollama_service(base_url, model_name):
return {"name": llm_name, "type": "llm", "errors": 1}
else:
if "api_key" in config and any(x in config["api_key"] for x in ["你的", "placeholder", "sk-xxx"]):
if "api_key" in config and any(
x in config["api_key"] for x in ["你的", "placeholder", "sk-xxx"]
):
print(f"🚫 跳过未配置的LLM: {llm_name}")
return {"name": llm_name, "type": "llm", "errors": 1}
# 获取实际类型（兼容旧配置）
module_type = config.get('type', llm_name)
module_type = config.get("type", llm_name)
llm = create_llm_instance(module_type, config)
# 统一使用UTF-8编码
test_sentences = [s.encode('utf-8').decode('utf-8') for s in self.test_sentences]
test_sentences = [
s.encode("utf-8").decode("utf-8") for s in self.test_sentences
]
# 创建所有句子的测试任务
sentence_tasks = []
for sentence in test_sentences:
sentence_tasks.append(self._test_single_sentence(llm_name, llm, sentence))
sentence_tasks.append(
self._test_single_sentence(llm_name, llm, sentence)
)
# 并发执行所有句子测试
sentence_results = await asyncio.gather(*sentence_tasks)
@@ -166,9 +171,15 @@ class AsyncPerformanceTester:
"type": "llm",
"avg_response": sum(response_times) / len(response_times),
"avg_first_token": sum(first_token_times) / len(first_token_times),
"std_first_token": statistics.stdev(first_token_times) if len(first_token_times) > 1 else 0,
"std_response": statistics.stdev(response_times) if len(response_times) > 1 else 0,
"errors": 0
"std_first_token": (
statistics.stdev(first_token_times)
if len(first_token_times) > 1
else 0
),
"std_response": (
statistics.stdev(response_times) if len(response_times) > 1 else 0
),
"errors": 0,
}
except Exception as e:
print(f"LLM {llm_name} 测试失败: {str(e)}")
@@ -184,8 +195,10 @@ class AsyncPerformanceTester:
async def process_response():
nonlocal first_token_received, first_token_time
for chunk in llm.response("perf_test", [{"role": "user", "content": sentence}]):
if not first_token_received and chunk.strip() != '':
for chunk in llm.response(
"perf_test", [{"role": "user", "content": sentence}]
):
if not first_token_received and chunk.strip() != "":
first_token_time = time.time() - sentence_start
first_token_received = True
print(f"✓ {llm_name} 首个Token: {first_token_time:.3f}s")
@@ -199,13 +212,15 @@ class AsyncPerformanceTester:
print(f"✓ {llm_name} 完成响应: {response_time:.3f}s")
if first_token_time is None:
first_token_time = response_time # 如果没有检测到first token，使用总响应时间
first_token_time = (
response_time # 如果没有检测到first token，使用总响应时间
)
return {
"name": llm_name,
"type": "llm",
"first_token_time": first_token_time,
"response_time": response_time
"response_time": response_time,
}
except Exception as e:
print(f"⚠️ {llm_name} 句子测试失败: {str(e)}")
@@ -214,24 +229,37 @@ class AsyncPerformanceTester:
def _generate_combinations(self):
"""生成最佳组合建议"""
valid_llms = [
k for k, v in self.results["llm"].items()
k
for k, v in self.results["llm"].items()
if v["errors"] == 0 and v["avg_first_token"] >= 0.05
]
valid_tts = [k for k, v in self.results["tts"].items() if v["errors"] == 0]
# 找出基准值
min_first_token = min([self.results["llm"][llm]["avg_first_token"] for llm in valid_llms]) if valid_llms else 1
min_tts_time = min([self.results["tts"][tts]["avg_time"] for tts in valid_tts]) if valid_tts else 1
min_first_token = (
min([self.results["llm"][llm]["avg_first_token"] for llm in valid_llms])
if valid_llms
else 1
)
min_tts_time = (
min([self.results["tts"][tts]["avg_time"] for tts in valid_tts])
if valid_tts
else 1
)
for llm in valid_llms:
for tts in valid_tts:
# 计算相对性能分数（越小越好）
llm_score = self.results["llm"][llm]["avg_first_token"] / min_first_token
llm_score = (
self.results["llm"][llm]["avg_first_token"] / min_first_token
)
tts_score = self.results["tts"][tts]["avg_time"] / min_tts_time
# 计算稳定性分数（标准差/平均值，越小越稳定）
llm_stability = self.results["llm"][llm]["std_first_token"] / self.results["llm"][llm][
"avg_first_token"]
llm_stability = (
self.results["llm"][llm]["std_first_token"]
/ self.results["llm"][llm]["avg_first_token"]
)
# 综合得分（考虑性能和稳定性）
# 性能权重0.7，稳定性权重0.3
@@ -240,16 +268,20 @@ class AsyncPerformanceTester:
# 总分 = LLM得分(70%) + TTS得分(30%)
total_score = llm_final_score * 0.7 + tts_score * 0.3
self.results["combinations"].append({
"llm": llm,
"tts": tts,
"score": total_score,
"details": {
"llm_first_token": self.results["llm"][llm]["avg_first_token"],
"llm_stability": llm_stability,
"tts_time": self.results["tts"][tts]["avg_time"]
self.results["combinations"].append(
{
"llm": llm,
"tts": tts,
"score": total_score,
"details": {
"llm_first_token": self.results["llm"][llm][
"avg_first_token"
],
"llm_stability": llm_stability,
"tts_time": self.results["tts"][tts]["avg_time"],
},
}
})
)
# 分数越小越好
self.results["combinations"].sort(key=lambda x: x["score"])
@@ -260,42 +292,45 @@ class AsyncPerformanceTester:
for name, data in self.results["llm"].items():
if data["errors"] == 0:
stability = data["std_first_token"] / data["avg_first_token"]
llm_table.append([
name, # 不需要固定宽度，让tabulate自己处理对齐
f"{data['avg_first_token']:.3f}秒",
f"{data['avg_response']:.3f}秒",
f"{stability:.3f}"
])
llm_table.append(
[
name, # 不需要固定宽度，让tabulate自己处理对齐
f"{data['avg_first_token']:.3f}秒",
f"{data['avg_response']:.3f}秒",
f"{stability:.3f}",
]
)
if llm_table:
print("\nLLM 性能排行:")
print(tabulate(
llm_table,
headers=["模型名称", "首字耗时", "总耗时", "稳定性"],
tablefmt="github",
colalign=("left", "right", "right", "right"),
disable_numparse=True
))
print(
tabulate(
llm_table,
headers=["模型名称", "首字耗时", "总耗时", "稳定性"],
tablefmt="github",
colalign=("left", "right", "right", "right"),
disable_numparse=True,
)
)
else:
print("\n⚠️ 没有可用的LLM模块进行测试。")
tts_table = []
for name, data in self.results["tts"].items():
if data["errors"] == 0:
tts_table.append([
name, # 不需要固定宽度
f"{data['avg_time']:.3f}秒"
])
tts_table.append([name, f"{data['avg_time']:.3f}秒"]) # 不需要固定宽度
if tts_table:
print("\nTTS 性能排行:")
print(tabulate(
tts_table,
headers=["模型名称", "合成耗时"],
tablefmt="github",
colalign=("left", "right"),
disable_numparse=True
))
print(
tabulate(
tts_table,
headers=["模型名称", "合成耗时"],
tablefmt="github",
colalign=("left", "right"),
disable_numparse=True,
)
)
else:
print("\n⚠️ 没有可用的TTS模块进行测试。")
@@ -303,21 +338,31 @@ class AsyncPerformanceTester:
print("\n推荐配置组合 (得分越小越好):")
combo_table = []
for combo in self.results["combinations"][:5]:
combo_table.append([
f"{combo['llm']} + {combo['tts']}", # 不需要固定宽度
f"{combo['score']:.3f}",
f"{combo['details']['llm_first_token']:.3f}秒",
f"{combo['details']['llm_stability']:.3f}",
f"{combo['details']['tts_time']:.3f}秒"
])
combo_table.append(
[
f"{combo['llm']} + {combo['tts']}", # 不需要固定宽度
f"{combo['score']:.3f}",
f"{combo['details']['llm_first_token']:.3f}秒",
f"{combo['details']['llm_stability']:.3f}",
f"{combo['details']['tts_time']:.3f}秒",
]
)
print(tabulate(
combo_table,
headers=["组合方案", "综合得分", "LLM首字耗时", "稳定性", "TTS合成耗时"],
tablefmt="github",
colalign=("left", "right", "right", "right", "right"),
disable_numparse=True
))
print(
tabulate(
combo_table,
headers=[
"组合方案",
"综合得分",
"LLM首字耗时",
"稳定性",
"TTS合成耗时",
],
tablefmt="github",
colalign=("left", "right", "right", "right", "right"),
disable_numparse=True,
)
)
else:
print("\n⚠️ 没有可用的模块组合建议。")
@@ -341,18 +386,21 @@ class AsyncPerformanceTester:
for llm_name, config in self.config.get("LLM", {}).items():
# 检查配置有效性
if llm_name == "CozeLLM":
if any(x in config.get("bot_id", "") for x in ["你的"]) \
or any(x in config.get("user_id", "") for x in ["你的"]):
if any(x in config.get("bot_id", "") for x in ["你的"]) or any(
x in config.get("user_id", "") for x in ["你的"]
):
print(f"⏭️ LLM {llm_name} 未配置bot_id/user_id，已跳过")
continue
elif "api_key" in config and any(x in config["api_key"] for x in ["你的", "placeholder", "sk-xxx"]):
elif "api_key" in config and any(
x in config["api_key"] for x in ["你的", "placeholder", "sk-xxx"]
):
print(f"⏭️ LLM {llm_name} 未配置api_key，已跳过")
continue
# 对于Ollama，先检查服务状态
if llm_name == "Ollama":
base_url = config.get('base_url', 'http://localhost:11434')
model_name = config.get('model_name')
base_url = config.get("base_url", "http://localhost:11434")
model_name = config.get("model_name")
if not model_name:
print(f"🚫 Ollama未配置model_name")
continue
@@ -361,27 +409,33 @@ class AsyncPerformanceTester:
continue
print(f"📋 添加LLM测试任务: {llm_name}")
module_type = config.get('type', llm_name)
module_type = config.get("type", llm_name)
llm = create_llm_instance(module_type, config)
# 为每个句子创建独立任务
for sentence in self.test_sentences:
sentence = sentence.encode('utf-8').decode('utf-8')
sentence = sentence.encode("utf-8").decode("utf-8")
all_tasks.append(self._test_single_sentence(llm_name, llm, sentence))
# TTS测试任务
for tts_name, config in self.config.get("TTS", {}).items():
token_fields = ["access_token", "api_key", "token"]
if any(field in config and any(x in config[field] for x in ["你的", "placeholder"]) for field in
token_fields):
if any(
field in config
and any(x in config[field] for x in ["你的", "placeholder"])
for field in token_fields
):
print(f"⏭️ TTS {tts_name} 未配置access_token/api_key，已跳过")
continue
print(f"🎵 添加TTS测试任务: {tts_name}")
all_tasks.append(self._test_tts(tts_name, config))
print(
f"\n✅ 找到 {len([t for t in all_tasks if 'test_single_sentence' in str(t)]) / len(self.test_sentences):.0f} 个可用LLM模块")
print(f"✅ 找到 {len([t for t in all_tasks if '_test_tts' in str(t)])} 个可用TTS模块")
f"\n✅ 找到 {len([t for t in all_tasks if 'test_single_sentence' in str(t)]) / len(self.test_sentences):.0f} 个可用LLM模块"
)
print(
f"✅ 找到 {len([t for t in all_tasks if '_test_tts' in str(t)])} 个可用TTS模块"
)
print("\n⏳ 开始并发测试所有模块...\n")
# 并发执行所有测试任务
@@ -389,7 +443,11 @@ class AsyncPerformanceTester:
# 处理LLM结果
llm_results = {}
for result in [r for r in all_results if r and isinstance(r, dict) and r.get("type") == "llm"]:
for result in [
r
for r in all_results
if r and isinstance(r, dict) and r.get("type") == "llm"
]:
llm_name = result["name"]
if llm_name not in llm_results:
llm_results[llm_name] = {
@@ -397,9 +455,11 @@ class AsyncPerformanceTester:
"type": "llm",
"first_token_times": [],
"response_times": [],
"errors": 0
"errors": 0,
}
llm_results[llm_name]["first_token_times"].append(result["first_token_time"])
llm_results[llm_name]["first_token_times"].append(
result["first_token_time"]
)
llm_results[llm_name]["response_times"].append(result["response_time"])
# 计算LLM平均值和标准差
@@ -408,16 +468,29 @@ class AsyncPerformanceTester:
self.results["llm"][llm_name] = {
"name": llm_name,
"type": "llm",
"avg_response": sum(data["response_times"]) / len(data["response_times"]),
"avg_first_token": sum(data["first_token_times"]) / len(data["first_token_times"]),
"std_first_token": statistics.stdev(data["first_token_times"]) if len(
data["first_token_times"]) > 1 else 0,
"std_response": statistics.stdev(data["response_times"]) if len(data["response_times"]) > 1 else 0,
"errors": 0
"avg_response": sum(data["response_times"])
/ len(data["response_times"]),
"avg_first_token": sum(data["first_token_times"])
/ len(data["first_token_times"]),
"std_first_token": (
statistics.stdev(data["first_token_times"])
if len(data["first_token_times"]) > 1
else 0
),
"std_response": (
statistics.stdev(data["response_times"])
if len(data["response_times"]) > 1
else 0
),
"errors": 0,
}
# 处理TTS结果
for result in [r for r in all_results if r and isinstance(r, dict) and r.get("type") == "tts"]:
for result in [
r
for r in all_results
if r and isinstance(r, dict) and r.get("type") == "tts"
]:
if result["errors"] == 0:
self.results["tts"][result["name"]] = result
@@ -433,4 +506,4 @@ async def main():
if __name__ == "__main__":
asyncio.run(main())
asyncio.run(main())