Files
life-echo/api/tests/test_llm_gateway.py
Kevin ac436b87a2 feat(api): 收敛对话与记忆流程边界,引入 LLM 网关与专用服务
- MemoryService 异步路径委托 MemoryIngestService / MemoryRetrievalService;富化派发经 MemoryEnrichmentScheduler
- WebSocket pipeline 经 ChatTurnService 与显式 DTO 编排单轮对话;回忆录片段入队由 MemoirIngestScheduler 封装
- 新增 LlmGateway(LlmUseCase),各 agent、任务与适配器对齐 ports
- 补充 memory 提示适配、runtime 类型、memory-retrieval 文档、ai-touchpoints 说明与扫描脚本及配套测试

Made-with: Cursor
2026-04-30 09:17:01 +08:00

63 lines
1.7 KiB
Python

from __future__ import annotations
import pytest
from app.core.llm_gateway import LlmGateway, LlmUseCase
class _FakeProvider:
def __init__(self, name: str) -> None:
self.name = name
self.langchain_llm = f"lc-{name}"
self.complete_calls: list[dict] = []
async def complete(self, messages, **kwargs) -> str:
self.complete_calls.append({"messages": messages, **kwargs})
return f"ok-{self.name}"
def test_llm_gateway_selects_default_or_fast_provider(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from app.core import llm_gateway as gateway_mod
default = _FakeProvider("default")
fast = _FakeProvider("fast")
monkeypatch.setattr(gateway_mod, "get_llm_provider", lambda: default)
monkeypatch.setattr(gateway_mod, "get_llm_provider_fast", lambda: fast)
gateway = LlmGateway()
assert gateway.langchain_llm_for() == "lc-default"
assert gateway.langchain_llm_for(LlmUseCase("memory", fast=True)) == "lc-fast"
@pytest.mark.asyncio
async def test_llm_gateway_chat_text_applies_use_case_defaults(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from app.core import llm_gateway as gateway_mod
provider = _FakeProvider("default")
monkeypatch.setattr(gateway_mod, "get_llm_provider", lambda: provider)
text = await LlmGateway().chat_text(
[{"role": "user", "content": "hi"}],
use_case=LlmUseCase(
"chat",
max_tokens=99,
temperature=0.2,
model="model-a",
),
)
assert text == "ok-default"
assert provider.complete_calls == [
{
"messages": [{"role": "user", "content": "hi"}],
"temperature": 0.2,
"model": "model-a",
"max_tokens": 99,
}
]