a84467c385
OpenRouter is used when no provider flag is set. Claude is selected with .cl, and GLM stays available via .glm.
286 lines
11 KiB
Python
286 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock
|
|
|
|
from micro_agents import Conversation, LlmMessage, Tool, ToolCallMessage, ToolCallResponseMessage, UserMessage
|
|
from micro_agents.conversation import ReasoningMessage
|
|
|
|
from app.llm import AnthropicMessagesLLM, ChatCompletionsLLM, OpenRouterChatLLM
|
|
|
|
|
|
class ChatCompletionsLlmTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_uses_chat_completions_with_system_prompt_and_thinking(self) -> None:
|
|
response_message = SimpleNamespace(
|
|
content="done",
|
|
reasoning_content="reasoning",
|
|
tool_calls=None,
|
|
)
|
|
create = AsyncMock(
|
|
return_value=SimpleNamespace(
|
|
choices=[SimpleNamespace(message=response_message)],
|
|
)
|
|
)
|
|
llm = ChatCompletionsLLM("https://glm.example/v4", "key", "glm-model")
|
|
llm._ai = SimpleNamespace( # type: ignore[assignment]
|
|
chat=SimpleNamespace(completions=SimpleNamespace(create=create))
|
|
)
|
|
|
|
result = await llm.ask(
|
|
Conversation([UserMessage("hello")]),
|
|
[],
|
|
instructions="system prompt",
|
|
thinking=True,
|
|
)
|
|
|
|
request = create.await_args.kwargs
|
|
self.assertEqual(request["model"], "glm-model")
|
|
self.assertEqual(
|
|
request["messages"],
|
|
[
|
|
{"role": "system", "content": "system prompt"},
|
|
{"role": "user", "content": "hello"},
|
|
],
|
|
)
|
|
self.assertEqual(request["extra_body"], {"thinking": {"type": "enabled"}})
|
|
self.assertIsInstance(result[0], ReasoningMessage)
|
|
self.assertIsInstance(result[1], LlmMessage)
|
|
|
|
async def test_serializes_and_parses_tool_calls(self) -> None:
|
|
model_call = SimpleNamespace(
|
|
id="call-2",
|
|
function=SimpleNamespace(name="lookup", arguments='{"query":"news"}'),
|
|
)
|
|
create = AsyncMock(
|
|
return_value=SimpleNamespace(
|
|
choices=[
|
|
SimpleNamespace(
|
|
message=SimpleNamespace(
|
|
content="",
|
|
reasoning_content=None,
|
|
tool_calls=[model_call],
|
|
)
|
|
)
|
|
],
|
|
)
|
|
)
|
|
llm = ChatCompletionsLLM("https://glm.example/v4", "key", "glm-model")
|
|
llm._ai = SimpleNamespace( # type: ignore[assignment]
|
|
chat=SimpleNamespace(completions=SimpleNamespace(create=create))
|
|
)
|
|
|
|
def lookup(query: str) -> str:
|
|
"""Look up a query."""
|
|
return query
|
|
|
|
conversation = Conversation(
|
|
[
|
|
UserMessage("find it"),
|
|
ReasoningMessage("need a tool"),
|
|
LlmMessage("checking"),
|
|
ToolCallMessage("call-1", "lookup", {"query": "weather"}),
|
|
ToolCallResponseMessage("call-1", "sunny"),
|
|
]
|
|
)
|
|
result = await llm.ask(conversation, [Tool(lookup)], thinking=False)
|
|
|
|
request = create.await_args.kwargs
|
|
self.assertEqual(request["extra_body"], {"thinking": {"type": "disabled"}})
|
|
self.assertEqual(request["tools"][0]["type"], "function")
|
|
self.assertEqual(request["tools"][0]["function"]["name"], "lookup")
|
|
self.assertEqual(request["messages"][1]["role"], "assistant")
|
|
self.assertEqual(request["messages"][1]["reasoning_content"], "need a tool")
|
|
self.assertEqual(request["messages"][2]["role"], "tool")
|
|
self.assertIsInstance(result[0], ToolCallMessage)
|
|
self.assertEqual(result[0].tool_name(), "lookup")
|
|
self.assertEqual(result[0].arguments(), {"query": "news"})
|
|
|
|
|
|
class OpenRouterChatLlmTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_requests_reasoning_effort_and_echoes_details(self) -> None:
|
|
details = [{"type": "reasoning.text", "text": "step", "signature": "sig"}]
|
|
create = AsyncMock(
|
|
return_value=SimpleNamespace(
|
|
choices=[
|
|
SimpleNamespace(
|
|
message=SimpleNamespace(
|
|
content="done",
|
|
reasoning="step",
|
|
reasoning_details=details,
|
|
tool_calls=None,
|
|
model_extra={},
|
|
)
|
|
)
|
|
]
|
|
)
|
|
)
|
|
llm = OpenRouterChatLLM("https://openrouter.ai/api/v1", "key", "anthropic/claude-haiku-5.5")
|
|
llm._ai = SimpleNamespace( # type: ignore[assignment]
|
|
chat=SimpleNamespace(completions=SimpleNamespace(create=create))
|
|
)
|
|
|
|
conversation = Conversation(
|
|
[
|
|
UserMessage("hello"),
|
|
ReasoningMessage("step", provider_item={"reasoning_details": details}),
|
|
LlmMessage("checking"),
|
|
ToolCallMessage("call-1", "lookup", {"query": "weather"}),
|
|
ToolCallResponseMessage("call-1", "sunny"),
|
|
]
|
|
)
|
|
result = await llm.ask(conversation, [], thinking=True)
|
|
|
|
request = create.await_args.kwargs
|
|
self.assertEqual(request["model"], "anthropic/claude-haiku-5.5")
|
|
self.assertEqual(request["extra_body"], {"reasoning": {"effort": "medium"}})
|
|
self.assertNotIn("thinking", request["extra_body"])
|
|
assistant = request["messages"][1]
|
|
self.assertEqual(assistant["reasoning"], "step")
|
|
self.assertEqual(assistant["reasoning_details"], details)
|
|
self.assertNotIn("reasoning_content", assistant)
|
|
self.assertIsInstance(result[0], ReasoningMessage)
|
|
self.assertEqual(result[0].provider_item(), {"reasoning_details": details})
|
|
self.assertEqual(result[1].message(), "done")
|
|
|
|
async def test_omits_reasoning_body_without_flag(self) -> None:
|
|
create = AsyncMock(
|
|
return_value=SimpleNamespace(
|
|
choices=[
|
|
SimpleNamespace(
|
|
message=SimpleNamespace(
|
|
content="ok",
|
|
reasoning=None,
|
|
reasoning_content=None,
|
|
reasoning_details=None,
|
|
tool_calls=None,
|
|
model_extra={},
|
|
)
|
|
)
|
|
]
|
|
)
|
|
)
|
|
llm = OpenRouterChatLLM("https://openrouter.ai/api/v1", "key", "anthropic/claude-haiku-5.5")
|
|
llm._ai = SimpleNamespace( # type: ignore[assignment]
|
|
chat=SimpleNamespace(completions=SimpleNamespace(create=create))
|
|
)
|
|
await llm.ask(Conversation([UserMessage("hello")]), [], thinking=False)
|
|
|
|
self.assertNotIn("extra_body", create.await_args.kwargs)
|
|
|
|
|
|
class AnthropicMessagesLlmTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_sends_messages_api_request_and_parses_blocks(self) -> None:
|
|
captured: list[dict[str, object]] = []
|
|
|
|
async def fake_post(payload: dict[str, object]) -> dict[str, object]:
|
|
captured.append(payload)
|
|
return {
|
|
"content": [
|
|
{
|
|
"type": "thinking",
|
|
"thinking": "look it up",
|
|
"signature": "sig",
|
|
},
|
|
{"type": "text", "text": "done"},
|
|
{
|
|
"type": "tool_use",
|
|
"id": "toolu_1",
|
|
"name": "lookup",
|
|
"input": {"query": "news"},
|
|
},
|
|
]
|
|
}
|
|
|
|
llm = AnthropicMessagesLLM("https://api.anthropic.com", "key", "claude-haiku-5-5")
|
|
llm._post = fake_post # type: ignore[method-assign]
|
|
|
|
def lookup(query: str) -> str:
|
|
"""Look up a query."""
|
|
return query
|
|
|
|
thinking = {
|
|
"type": "thinking",
|
|
"thinking": "previous",
|
|
"signature": "old-sig",
|
|
}
|
|
result = await llm.ask(
|
|
Conversation(
|
|
[
|
|
UserMessage("hello", media=[]),
|
|
ReasoningMessage("previous", provider_item=thinking),
|
|
LlmMessage("checking"),
|
|
ToolCallMessage("toolu_0", "lookup", {"query": "weather"}),
|
|
ToolCallResponseMessage("toolu_0", "sunny"),
|
|
]
|
|
),
|
|
[Tool(lookup)],
|
|
instructions="system prompt",
|
|
thinking=True,
|
|
)
|
|
|
|
payload = captured[0]
|
|
self.assertEqual(llm._messages_url(), "https://api.anthropic.com/v1/messages")
|
|
self.assertEqual(payload["model"], "claude-haiku-5-5")
|
|
self.assertEqual(payload["system"], "system prompt")
|
|
self.assertEqual(payload["thinking"], {"type": "adaptive", "display": "summarized"})
|
|
self.assertEqual(payload["tools"][0]["name"], "lookup")
|
|
self.assertEqual(payload["tools"][0]["input_schema"]["type"], "object")
|
|
self.assertEqual(
|
|
payload["messages"],
|
|
[
|
|
{"role": "user", "content": [{"type": "text", "text": "hello"}]},
|
|
{
|
|
"role": "assistant",
|
|
"content": [
|
|
thinking,
|
|
{"type": "text", "text": "checking"},
|
|
{
|
|
"type": "tool_use",
|
|
"id": "toolu_0",
|
|
"name": "lookup",
|
|
"input": {"query": "weather"},
|
|
},
|
|
],
|
|
},
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{
|
|
"type": "tool_result",
|
|
"tool_use_id": "toolu_0",
|
|
"content": "sunny",
|
|
}
|
|
],
|
|
},
|
|
],
|
|
)
|
|
self.assertIsInstance(result[0], ReasoningMessage)
|
|
self.assertEqual(
|
|
result[0].provider_item(),
|
|
{"type": "thinking", "thinking": "look it up", "signature": "sig"},
|
|
)
|
|
self.assertIsInstance(result[1], LlmMessage)
|
|
self.assertEqual(result[1].message(), "done")
|
|
self.assertIsInstance(result[2], ToolCallMessage)
|
|
self.assertEqual(result[2].arguments(), {"query": "news"})
|
|
|
|
async def test_disables_thinking_without_flag(self) -> None:
|
|
captured: list[dict[str, object]] = []
|
|
|
|
async def fake_post(payload: dict[str, object]) -> dict[str, object]:
|
|
captured.append(payload)
|
|
return {"content": [{"type": "text", "text": "ok"}]}
|
|
|
|
llm = AnthropicMessagesLLM("https://api.anthropic.com/v1", "key", "claude-haiku-5-5")
|
|
llm._post = fake_post # type: ignore[method-assign]
|
|
await llm.ask(Conversation([UserMessage("hello")]), [], thinking=False)
|
|
|
|
self.assertEqual(llm._messages_url(), "https://api.anthropic.com/v1/messages")
|
|
self.assertEqual(captured[0]["thinking"], {"type": "disabled"})
|
|
self.assertNotIn("tools", captured[0])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|