#!/usr/bin/env python3 """LLM 抽象层测试。 重点验证实测踩过的三个坑是否真的被吸收： 1. 思考字段名差异（reasoning / reasoning_content） 2. 关闭思考的方式差异 3. 多条 system 消息归一化 运行： ~/venvs/ai_assistant/bin/python -m app.tests.test_llm 或 cd <项目根> && ~/venvs/ai_assistant/bin/python app/tests/test_llm.py """ from __future__ import annotations import asyncio import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent)) from app.config import AppConfig # noqa: E402 from app.llm import ( # noqa: E402 ChunkKind, DeepSeekProvider, FallbackProvider, LocalVLLMProvider, create_provider, ) from app.llm.base import LLMProvider # noqa: E402 OK, FAIL = [], [] def check(name: str, cond: bool, detail: str = "") -> None: (OK if cond else FAIL).append(name) print(f" [{'PASS' if cond else 'FAIL'}] {name}" + (f" — {detail}" if detail else "")) # ---------------------------------------------------------------- 纯本地逻辑（无需网络） def test_normalize_messages() -> None: print("\n=== 1. 消息归一化（多条 system → 一条，置于开头）===") msgs = [ {"role": "system", "content": "你是助手。"}, {"role": "user", "content": "你好"}, {"role": "system", "content": "记忆：用户叫 ylhan。"}, ] out = LLMProvider.normalize_messages(msgs) check("只剩一条 system", sum(1 for m in out if m["role"] == "system") == 1, f"实际 {sum(1 for m in out if m['role']=='system')} 条") check("system 在开头", out[0]["role"] == "system") check("两条 system 内容都保留", "你是助手。" in out[0]["content"] and "ylhan" in out[0]["content"]) check("非 system 消息顺序不变", [m["role"] for m in out[1:]] == ["user"]) # 无 system 的情况不应凭空造一条 out2 = LLMProvider.normalize_messages([{"role": "user", "content": "hi"}]) check("无 system 时不新增", len(out2) == 1 and out2[0]["role"] == "user") # 空 system 应被丢弃 out3 = LLMProvider.normalize_messages( [{"role": "system", "content": " "}, {"role": "user", "content": "hi"}]) check("空白 system 被丢弃", len(out3) == 1) def test_request_bodies() -> None: print("\n=== 2. 各后端的请求体（关思考方式不同）===") cfg_local = AppConfig.load().llm cfg_ds = AppConfig.load().llm_fallback cfg_local.thinking = False cfg_ds.thinking = False b_local = LocalVLLMProvider(cfg_local)._build_body( [{"role": "user", "content": "hi"}], tools=None, max_tokens=100) check("本机用 chat_template_kwargs", b_local.get("chat_template_kwargs") == {"enable_thinking": False}, str(b_local.get("chat_template_kwargs"))) b_ds = DeepSeekProvider(cfg_ds)._build_body( [{"role": "user", "content": "hi"}], tools=None, max_tokens=100) check("DeepSeek 用 thinking.type", b_ds.get("thinking") == {"type": "disabled"}, str(b_ds.get("thinking"))) check("两者模型名不同", b_local["model"] != b_ds["model"], f"{b_local['model']} vs {b_ds['model']}") def test_sse_parsing() -> None: print("\n=== 3. SSE 解析：两种思考字段名都认 ===") class FakeResp: def __init__(self, lines): self.content = self._gen(lines) async def _gen(self, lines): for l in lines: yield (l + "\n").encode() async def run(lines): return [c async for c in LLMProvider._iter_sse(FakeResp(lines))] async def main(): # reasoning_content（DeepSeek 写法） c1 = await run([ 'data: {"choices":[{"delta":{"reasoning_content":"想一想"}}]}', 'data: {"choices":[{"delta":{"content":"你好"}}]}', 'data: [DONE]', ]) check("认 reasoning_content", any(c.kind == ChunkKind.REASONING and c.text == "想一想" for c in c1)) check("content 正常", any(c.kind == ChunkKind.CONTENT and c.text == "你好" for c in c1)) # reasoning（本机 vLLM 写法） c2 = await run([ 'data: {"choices":[{"delta":{"reasoning":"想一下"}}]}', 'data: {"choices":[{"delta":{"content":"在的"}}]}', 'data: [DONE]', ]) check("认 reasoning（本机 vLLM 字段）", any(c.kind == ChunkKind.REASONING and c.text == "想一下" for c in c2)) # 工具调用分片累积 c3 = await run([ 'data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"c1","function":{"name":"set_light","arguments":"{\\"ro"}}]}}]}', 'data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"om\\":\\"客厅\\"}"}}]}}]}', 'data: {"choices":[{"delta":{},"finish_reason":"tool_calls"}]}', ]) tc = [c for c in c3 if c.kind == ChunkKind.TOOL_CALL] check("工具调用参数分片拼接", tc and tc[-1].tool_arguments == '{"room":"客厅"}', tc[-1].tool_arguments if tc else "无") check("工具名保留", tc and tc[-1].tool_name == "set_light") # 坏帧不应崩溃 c4 = await run(['data: {坏json', 'data: {"choices":[{"delta":{"content":"ok"}}]}', 'data: [DONE]']) check("坏帧被跳过而非崩溃", any(c.kind == ChunkKind.CONTENT and c.text == "ok" for c in c4)) asyncio.run(main()) # ---------------------------------------------------------------- 需要网络 async def test_live() -> None: cfg = AppConfig.load() provider = create_provider(cfg.llm) print(f"\n=== 4. 实连本机 vLLM（{cfg.llm.model} @ {cfg.llm.base_url}）===") ok, detail = await provider.health() check("健康检查", ok, detail) if not ok: await provider.aclose() return msgs = [ {"role": "system", "content": "你是桌面陪伴助手，回复必须简短自然、适合朗读，不超过两句话，" "不要用 Markdown 或表情符号。"}, {"role": "user", "content": "今天天气怎么样"}, {"role": "assistant", "content": "今天北京晴，十八到二十六度。"}, {"role": "user", "content": "那明天呢"}, ] res = await provider.complete(msgs, max_tokens=120) print(f" 回答: {res.content.strip()[:70]!r}") print(f" 首token(回答)={res.first_content_ms:.0f}ms 总={res.total_ms:.0f}ms") print(f" 思考={len(res.reasoning)}字 完成tok={res.usage.get('completion_tokens') if res.usage else '?'}") check("有回答内容", bool(res.content.strip())) check("关思考后无思考内容", len(res.reasoning) == 0, f"{len(res.reasoning)} 字") check("首token在 1 秒内", (res.first_content_ms or 9e9) < 1000, f"{res.first_content_ms:.0f}ms") check("回复足够短", len(res.content) < 100, f"{len(res.content)} 字") # 工具调用 tools = [{"type": "function", "function": { "name": "set_light", "description": "开关指定房间的灯", "parameters": {"type": "object", "properties": { "room": {"type": "string"}, "on": {"type": "boolean"}}, "required": ["room", "on"]}}}] res2 = await provider.complete([{"role": "user", "content": "把客厅的灯打开"}], tools=tools, max_tokens=128) check("工具调用正确", res2.has_tool_calls and res2.tool_calls[0].parsed_arguments().get("room") == "客厅", f"{res2.tool_calls[0].name if res2.tool_calls else '无'}") # 取消：模拟打断 async def cancel_midway(): async for _ch in provider.stream( [{"role": "user", "content": "请详细讲一下中国历史"}], max_tokens=800): return "cancelled" return "finished" got = await asyncio.wait_for(cancel_midway(), timeout=30) check("可中途取消（打断）", got == "cancelled", got) await provider.aclose() async def test_fallback_logic() -> None: print("\n=== 5. 降级逻辑（不发真实请求）===") cfg = AppConfig.load() class BoomProvider(LocalVLLMProvider): def __init__(self, c, err): super().__init__(c); self._err = err async def stream(self, *a, **kw): raise self._err yield # noqa: 使其成为生成器 fallback = DeepSeekProvider(cfg.llm_fallback) # 连接类错误应降级 from app.llm.base import LLMError p = FallbackProvider(BoomProvider(cfg.llm, ConnectionError("连不上")), fallback) check("连接错误判为可降级", p._should_fallback(ConnectionError("x"))) check("超时判为可降级", p._should_fallback(TimeoutError("x"))) check("503 判为可降级", p._should_fallback(LLMError("local HTTP 503: busy"))) check("400 判为不可降级", not p._should_fallback(LLMError("local HTTP 400: bad"))) check("401 判为不可降级", not p._should_fallback(LLMError("local HTTP 401: auth"))) def main() -> int: test_normalize_messages() test_request_bodies() test_sse_parsing() asyncio.run(test_live()) asyncio.run(test_fallback_logic()) print("\n" + "=" * 62) print(f"通过 {len(OK)} / {len(OK) + len(FAIL)}") for n in FAIL: print(f" 未通过: {n}") return 0 if not FAIL else 1 if __name__ == "__main__": sys.exit(main())