334 lines
13 KiB
Python
334 lines
13 KiB
Python
"""Tests for LLM response parsing - MiniMax/OpenAI format"""
|
||
import json
|
||
import pytest
|
||
from luxx.services.llm_response import ParsedDelta
|
||
|
||
|
||
class TestMiniMaxParsing:
|
||
"""Tests for MiniMax/OpenAI streaming format parsing"""
|
||
|
||
def test_parse_text_chunk(self):
|
||
"""Parse text content chunk"""
|
||
chunk_str = 'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"Hello","role":"assistant"}}]}'
|
||
chunk = json.loads(chunk_str[6:]) # Remove "data: "
|
||
|
||
delta = chunk["choices"][0]["delta"]
|
||
assert delta["content"] == "Hello"
|
||
|
||
def test_parse_text_accumulation(self):
|
||
"""Multiple chunks should accumulate to full text"""
|
||
chunks = [
|
||
'{"id":"msg_001","choices":[{"index":0,"delta":{"content":"你好","role":"assistant"}}]}',
|
||
'{"id":"msg_001","choices":[{"index":0,"delta":{"content":"！","role":"assistant"}}]}',
|
||
'{"id":"msg_001","choices":[{"index":0,"delta":{"content":"有什么","role":"assistant"}}]}',
|
||
'{"id":"msg_001","choices":[{"index":0,"delta":{"content":"可以帮助你的吗？","role":"assistant"}}]}',
|
||
]
|
||
|
||
full_text = ""
|
||
for c in chunks:
|
||
chunk = json.loads(c)
|
||
content = chunk["choices"][0]["delta"].get("content", "")
|
||
full_text += content
|
||
|
||
assert full_text == "你好！有什么可以帮助你的吗？"
|
||
|
||
def test_parse_finish_chunk(self):
|
||
"""Parse finish_reason chunk"""
|
||
chunk_str = 'data: {"id":"msg_001","choices":[{"finish_reason":"stop","index":0,"delta":{"role":"assistant"}}]}'
|
||
chunk = json.loads(chunk_str[6:])
|
||
|
||
finish_reason = chunk["choices"][0].get("finish_reason")
|
||
assert finish_reason == "stop"
|
||
|
||
def test_parse_usage_chunk(self):
|
||
"""Parse usage information"""
|
||
chunk_str = '{"id":"msg_001","usage":{"prompt_tokens":100,"completion_tokens":50,"total_tokens":150}}'
|
||
chunk = json.loads(chunk_str)
|
||
|
||
usage = chunk.get("usage", {})
|
||
assert usage["prompt_tokens"] == 100
|
||
assert usage["completion_tokens"] == 50
|
||
assert usage["total_tokens"] == 150
|
||
|
||
def test_parse_empty_content_chunk(self):
|
||
"""Parse chunk with empty content (just role)"""
|
||
chunk_str = 'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"","role":"assistant"}}]}'
|
||
chunk = json.loads(chunk_str[6:])
|
||
|
||
delta = chunk["choices"][0]["delta"]
|
||
content = delta.get("content", "")
|
||
assert content == ""
|
||
|
||
def test_parse_done_marker(self):
|
||
"""Parse [DONE] marker"""
|
||
chunk_str = "data: [DONE]"
|
||
assert chunk_str.strip().startswith("data: [DONE]")
|
||
|
||
|
||
class TestOpenAIAdapter:
|
||
"""Tests for OpenAI adapter parsing logic"""
|
||
|
||
@pytest.fixture
|
||
def adapter(self):
|
||
from luxx.services.llm_adapters.openai_adapter import OpenAIAdapter
|
||
return OpenAIAdapter()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_parse_stream_text(self, adapter):
|
||
"""Should parse text content"""
|
||
chunk = 'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"Hello"}}]}'
|
||
deltas = [d async for d in adapter.parse_stream_chunk(chunk)]
|
||
|
||
assert len(deltas) == 1
|
||
assert deltas[0].text == "Hello"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_parse_stream_finish(self, adapter):
|
||
"""Should detect completion"""
|
||
chunk = 'data: {"id":"msg_001","choices":[{"finish_reason":"stop","index":0,"delta":{}}]}'
|
||
deltas = [d async for d in adapter.parse_stream_chunk(chunk)]
|
||
|
||
assert len(deltas) == 1
|
||
assert deltas[0].is_complete is True
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_parse_stream_empty_content(self, adapter):
|
||
"""Should skip empty content chunks"""
|
||
chunk = 'data: {"id":"msg_001","choices":[{"index":0,"delta":{"role":"assistant"}}]}'
|
||
deltas = [d async for d in adapter.parse_stream_chunk(chunk)]
|
||
|
||
# Empty content without finish_reason should be skipped
|
||
assert len(deltas) == 0
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_parse_stream_done_marker(self, adapter):
|
||
"""Should handle [DONE] marker"""
|
||
chunk = "data: [DONE]"
|
||
deltas = [d async for d in adapter.parse_stream_chunk(chunk)]
|
||
|
||
assert len(deltas) == 1
|
||
assert deltas[0].is_complete is True
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_parse_stream_invalid_json(self, adapter):
|
||
"""Should handle invalid JSON gracefully"""
|
||
chunk = "not valid json"
|
||
deltas = [d async for d in adapter.parse_stream_chunk(chunk)]
|
||
|
||
assert len(deltas) == 0
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_parse_stream_empty_chunk(self, adapter):
|
||
"""Should handle empty chunk"""
|
||
deltas = [d async for d in adapter.parse_stream_chunk("")]
|
||
assert len(deltas) == 0
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_parse_stream_whitespace_chunk(self, adapter):
|
||
"""Should handle whitespace-only chunk"""
|
||
deltas = [d async for d in adapter.parse_stream_chunk(" \n")]
|
||
assert len(deltas) == 0
|
||
|
||
|
||
class TestBuildRequest:
|
||
"""Tests for request building"""
|
||
|
||
@pytest.fixture
|
||
def adapter(self):
|
||
from luxx.services.llm_adapters.openai_adapter import OpenAIAdapter
|
||
return OpenAIAdapter()
|
||
|
||
def test_build_request_basic(self, adapter):
|
||
"""Should build basic request"""
|
||
body, headers = adapter.build_request(
|
||
model="MiniMax-M2.5",
|
||
messages=[{"role": "user", "content": "Hello"}]
|
||
)
|
||
|
||
assert body["model"] == "MiniMax-M2.5"
|
||
assert body["messages"] == [{"role": "user", "content": "Hello"}]
|
||
assert body["stream"] is True
|
||
|
||
def test_build_request_with_tools(self, adapter):
|
||
"""Should include tools in request"""
|
||
tools = [{"type": "function", "function": {"name": "test", "parameters": {}}}]
|
||
body, _ = adapter.build_request(
|
||
model="MiniMax-M2.5",
|
||
messages=[],
|
||
tools=tools
|
||
)
|
||
|
||
assert "tools" in body
|
||
assert body["tool_choice"] == "auto"
|
||
|
||
def test_build_request_with_temperature(self, adapter):
|
||
"""Should include temperature"""
|
||
body, _ = adapter.build_request(
|
||
model="MiniMax-M2.5",
|
||
messages=[],
|
||
temperature=0.7
|
||
)
|
||
|
||
assert body["temperature"] == 0.7
|
||
|
||
def test_build_request_with_max_tokens(self, adapter):
|
||
"""Should include max_tokens"""
|
||
body, _ = adapter.build_request(
|
||
model="MiniMax-M2.5",
|
||
messages=[],
|
||
max_tokens=1000
|
||
)
|
||
|
||
assert body["max_tokens"] == 1000
|
||
|
||
|
||
class TestParsedDelta:
|
||
"""Tests for ParsedDelta dataclass"""
|
||
|
||
def test_default_values(self):
|
||
"""Should have correct defaults"""
|
||
delta = ParsedDelta()
|
||
assert delta.text == ""
|
||
assert delta.thinking == ""
|
||
assert delta.tool_call is None
|
||
assert delta.usage == {}
|
||
assert delta.is_complete is False
|
||
|
||
def test_with_text(self):
|
||
"""Should accept text content"""
|
||
delta = ParsedDelta(text="Hello world")
|
||
assert delta.text == "Hello world"
|
||
|
||
def test_with_usage(self):
|
||
"""Should accept usage dict"""
|
||
delta = ParsedDelta(usage={"prompt_tokens": 10, "completion_tokens": 5})
|
||
assert delta.usage["prompt_tokens"] == 10
|
||
|
||
def test_with_complete_flag(self):
|
||
"""Should accept is_complete flag"""
|
||
delta = ParsedDelta(is_complete=True)
|
||
assert delta.is_complete is True
|
||
|
||
|
||
class TestEndToEndStreaming:
|
||
"""End-to-end streaming simulation tests"""
|
||
|
||
@pytest.fixture
|
||
def adapter(self):
|
||
from luxx.services.llm_adapters.openai_adapter import OpenAIAdapter
|
||
return OpenAIAdapter()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_full_text_stream(self, adapter):
|
||
"""Simulate full text response stream"""
|
||
chunks = [
|
||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"用户","role":"assistant"}}]}',
|
||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"用中文","role":"assistant"}}]}',
|
||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"说","role":"assistant"}}]}',
|
||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"\"你好\"","role":"assistant"}}]}',
|
||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"，这是","role":"assistant"}}]}',
|
||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"一个简单的","role":"assistant"}}]}',
|
||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"问候。","role":"assistant"}}]}',
|
||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"\n","role":"assistant"}}]}',
|
||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"我应该","role":"assistant"}}]}',
|
||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"用中文","role":"assistant"}}]}',
|
||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"友好地","role":"assistant"}}]}',
|
||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"回应。","role":"assistant"}}]}',
|
||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"\n\n","role":"assistant"}}]}',
|
||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"","role":"assistant"}}]}',
|
||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"\n\n你好！","role":"assistant"}}]}',
|
||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"有什么","role":"assistant"}}]}',
|
||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"我可以","role":"assistant"}}]}',
|
||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"帮助你的吗？","role":"assistant"}}]}',
|
||
'data: {"id":"msg_001","choices":[{"finish_reason":"stop","index":0,"delta":{"role":"assistant"}}]}',
|
||
]
|
||
|
||
full_text = ""
|
||
is_complete = False
|
||
|
||
for chunk in chunks:
|
||
deltas = [d async for d in adapter.parse_stream_chunk(chunk)]
|
||
for delta in deltas:
|
||
if delta.is_complete:
|
||
is_complete = True
|
||
if delta.text:
|
||
full_text += delta.text
|
||
|
||
expected = "用户用中文说\"你好\"，这是一个简单的问候。我应该用中文友好地回应。\n\n你好！有什么我可以帮助你的吗？"
|
||
assert full_text == expected
|
||
assert is_complete is True
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_empty_stream_between_finish(self, adapter):
|
||
"""Handle empty content chunks before finish"""
|
||
chunks = [
|
||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"Hello","role":"assistant"}}]}',
|
||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"","role":"assistant"}}]}',
|
||
'data: {"id":"msg_001","choices":[{"index":0,"delta":{"content":"","role":"assistant"}}]}',
|
||
'data: {"id":"msg_001","choices":[{"finish_reason":"stop","index":0,"delta":{"role":"assistant"}}]}',
|
||
]
|
||
|
||
full_text = ""
|
||
for chunk in chunks:
|
||
deltas = [d async for d in adapter.parse_stream_chunk(chunk)]
|
||
for delta in deltas:
|
||
if delta.text:
|
||
full_text += delta.text
|
||
|
||
assert full_text == "Hello"
|
||
|
||
|
||
class TestToolCallParsing:
|
||
"""Tests for tool call parsing"""
|
||
|
||
@pytest.fixture
|
||
def adapter(self):
|
||
from luxx.services.llm_adapters.openai_adapter import OpenAIAdapter
|
||
return OpenAIAdapter()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_tool_call_chunk(self, adapter):
|
||
"""Parse tool call chunk"""
|
||
chunk = json.dumps({
|
||
"id": "chatcmpl_001",
|
||
"choices": [{
|
||
"index": 0,
|
||
"delta": {
|
||
"tool_calls": [{
|
||
"index": 0,
|
||
"id": "call_abc123",
|
||
"type": "function",
|
||
"function": {
|
||
"name": "web_search",
|
||
"arguments": '{"query":'
|
||
}
|
||
}]
|
||
}
|
||
}]
|
||
})
|
||
|
||
# The current adapter doesn't yield tool_call deltas
|
||
# This test documents current behavior
|
||
deltas = [d async for d in adapter.parse_stream_chunk(chunk)]
|
||
# Content is empty, no delta yielded
|
||
assert len(deltas) == 0
|
||
|
||
def test_tool_call_accumulation(self):
|
||
"""Simulate tool call argument accumulation"""
|
||
chunks = [
|
||
{"function": {"name": "python_eval", "arguments": "{"}},
|
||
{"function": {"name": "", "arguments": '"expr": '}},
|
||
{"function": {"name": "", "arguments": '"1 + 1"'}},
|
||
{"function": {"name": "", "arguments": "}"}},
|
||
]
|
||
|
||
accumulated = {"name": "", "arguments": ""}
|
||
for c in chunks:
|
||
if c["function"]["name"]:
|
||
accumulated["name"] += c["function"]["name"]
|
||
if c["function"]["arguments"]:
|
||
accumulated["arguments"] += c["function"]["arguments"]
|
||
|
||
assert accumulated["name"] == "python_eval"
|
||
assert accumulated["arguments"] == '{"expr": "1 + 1"}'
|