"""Library for LLM tool-calling integration tests. Provides keywords that create an Executor with a mock chat model that returns tool_calls, then execute messages and verify that tools were invoked correctly — without requiring a real LLM API key. """ from __future__ import annotations import asyncio from typing import Any from unittest.mock import MagicMock, patch from langchain_core.messages import AIMessage from cleveractors.result import ActorResult from cleveractors.runtime import Executor, create_executor class ToolCallingTestLib: """Keywords for integration-style tests of LLM tool-calling.""" def __init__(self) -> None: self._executor: Executor | None = None self._last_result: ActorResult | None = None self._mock_ainvoke_calls: list[dict[str, Any]] = [] self._patches: list[Any] = [] self._multi_round_expected_prompt: int = 0 def _teardown_patches(self) -> None: for p in self._patches: p.stop() self._patches.clear() def create_executor_with_tool_calling_agent( self, tool_name: str = "echo", message: str = "" ) -> None: """Create an Executor whose LLM agent is configured with a tool. The chat model is mocked so that ``ainvoke`` returns an AIMessage with tool_calls. The model first returns a tool-call message, then a final plain-text answer. """ self._teardown_patches() self._mock_ainvoke_calls = [] final_response_text = "Final answer after tool use" def _build_mock_model(*args, **kwargs): mock_model = MagicMock() mock_model.temperature = 0.7 async def _mock_ainvoke(messages, **invoke_kwargs): self._mock_ainvoke_calls.append({"kwargs": invoke_kwargs}) call_count = len(self._mock_ainvoke_calls) if call_count == 1 and invoke_kwargs.get("tools"): tool_content = message or "hello from tool" return AIMessage( content="", usage_metadata={ "input_tokens": 50, "output_tokens": 10, "total_tokens": 60, }, tool_calls=[ { "id": "call_001", "name": tool_name, "args": {"message": tool_content}, }, ], ) return AIMessage( content=final_response_text, usage_metadata={ "input_tokens": 30, "output_tokens": 15, "total_tokens": 45, }, ) async def _mock_astream(messages, **invoke_kwargs): self._mock_ainvoke_calls.append({"kwargs": invoke_kwargs}) call_count = len(self._mock_ainvoke_calls) if call_count == 1 and invoke_kwargs.get("tools"): tool_content = message or "hello from stream" yield AIMessage( content="tool_call_response", usage_metadata={ "input_tokens": 40, "output_tokens": 10, "total_tokens": 50, }, tool_calls=[ { "id": "call_001", "name": tool_name, "args": {"message": tool_content}, }, ], ) else: yield AIMessage( content=final_response_text, usage_metadata={ "input_tokens": 25, "output_tokens": 12, "total_tokens": 37, }, ) mock_model.ainvoke = _mock_ainvoke mock_model.astream = _mock_astream return mock_model patcher = patch( "cleveractors.agents.llm.build_chat_model", side_effect=_build_mock_model, ) patcher.start() self._patches.append(patcher) config = { "type": "llm", "name": "tool_test_agent", "provider": "openai", "model": "gpt-3.5-turbo", "config": { "tools": [{"name": tool_name}], }, } self._executor = create_executor( config_dict=config, credentials={"openai": {"api_key": "mock-key"}}, limits={}, pricing={}, ) async def _execute_async(self, message: str) -> ActorResult: return await self._executor.execute(message) def execute_with_message(self, message: str) -> None: """Execute the executor with *message* and store the result.""" try: self._last_result = asyncio.run(self._execute_async(message)) finally: self._teardown_patches() def execute_stream_with_message(self, message: str) -> None: """Execute via execute_stream and store the last result.""" async def _stream(): chunks = [] async for token in self._executor.execute_stream(message): chunks.append(token) return "".join(chunks) try: asyncio.run(_stream()) self._last_result = self._executor.last_result finally: self._teardown_patches() def result_is_valid_actor_result(self) -> None: assert isinstance(self._last_result, ActorResult), ( f"Expected ActorResult, got {type(self._last_result)}" ) def result_response_contains(self, text: str) -> None: assert self._last_result is not None assert text in self._last_result.response, ( f"Expected '{text}' in response, got: {self._last_result.response[:200]}" ) def result_has_nodes(self) -> None: assert self._last_result is not None assert len(self._last_result.nodes) > 0, "Expected at least one node in result" def result_prompt_tokens_greater_than_zero(self) -> None: assert self._last_result is not None assert self._last_result.prompt_tokens > 0, ( f"Expected prompt_tokens > 0, got {self._last_result.prompt_tokens}" ) def chat_model_was_invoked_with_tools(self) -> None: for call in self._mock_ainvoke_calls: if call.get("kwargs", {}).get("tools"): return raise AssertionError("chat_model.ainvoke was never called with tools keyword") def multiple_tool_rounds_occurred(self) -> None: assert len(self._mock_ainvoke_calls) >= 2, ( f"Expected >= 2 ainvoke calls, got {len(self._mock_ainvoke_calls)}" ) def tool_call_result_is_in_response(self) -> None: assert self._last_result is not None assert len(self._last_result.response.strip()) > 0, ( "Expected non-empty response after tool calls" ) # ------------------------------------------------------------------ # New keywords for issue #67 (stream_message tool-loop integration) # ------------------------------------------------------------------ def stream_result_response_is_non_empty(self) -> None: # noqa: D401 """Verify that the stream response is non empty.""" assert self._last_result is not None assert len(self._last_result.response.strip()) > 0, ( f"Expected non-empty stream response, got: {self._last_result.response!r}" ) def stream_result_prompt_tokens_greater_than_zero(self) -> None: """Verify stream result has non-zero accumulated prompt tokens.""" assert self._last_result is not None assert self._last_result.prompt_tokens > 0, ( f"Expected stream prompt_tokens > 0, got {self._last_result.prompt_tokens}" ) def stream_result_response_contains(self, text: str) -> None: """Verify the stream response contains *text*.""" assert self._last_result is not None assert text in self._last_result.response, ( f"Expected {text!r} in stream response, got: {self._last_result.response[:200]!r}" ) def stream_result_prompt_tokens_equals_multi_round_sum(self) -> None: """Verify accumulated prompt tokens equal the sum across all mock rounds.""" assert self._last_result is not None expected = self._multi_round_expected_prompt actual = self._last_result.prompt_tokens assert actual == expected, ( f"Expected prompt_tokens={expected} (sum of all rounds), got {actual}" ) def create_executor_with_multi_round_tool_calling_agent( self, tool_name: str = "echo", rounds: int = 2 ) -> None: """Create an Executor whose mock model returns tool_calls for *rounds* rounds then a final answer. Tracks per-round token counts for assertion. """ self._teardown_patches() self._mock_ainvoke_calls = [] # Per-round prompt tokens: 50 per tool round + 30 for final answer tool_round_prompt = 50 final_prompt = 30 tool_round_completion = 10 final_completion = 15 total_rounds = int(rounds) self._multi_round_expected_prompt = ( tool_round_prompt * total_rounds + final_prompt ) final_response_text = "Multi round final answer" def _build_mock_model(*args, **kwargs): # type: ignore[no-untyped-def] mock_model = MagicMock() mock_model.temperature = 0.7 async def _mock_ainvoke(messages, **invoke_kwargs): # type: ignore[no-untyped-def] self._mock_ainvoke_calls.append({"kwargs": invoke_kwargs}) call_count = len(self._mock_ainvoke_calls) if call_count <= total_rounds and invoke_kwargs.get("tools"): return AIMessage( content="", usage_metadata={ "input_tokens": tool_round_prompt, "output_tokens": tool_round_completion, "total_tokens": tool_round_prompt + tool_round_completion, }, tool_calls=[ { "id": f"call_{call_count:03d}", "name": tool_name, "args": {"message": f"round_{call_count}"}, } ], ) return AIMessage( content=final_response_text, usage_metadata={ "input_tokens": final_prompt, "output_tokens": final_completion, "total_tokens": final_prompt + final_completion, }, ) mock_model.ainvoke = _mock_ainvoke mock_model.astream = MagicMock() # not called in tool path return mock_model patcher = patch( "cleveractors.agents.llm.build_chat_model", side_effect=_build_mock_model, ) patcher.start() self._patches.append(patcher) config = { "type": "llm", "name": "multi_round_tool_agent", "provider": "openai", "model": "gpt-3.5-turbo", "config": { "tools": [{"name": tool_name}], }, } self._executor = create_executor( config_dict=config, credentials={"openai": {"api_key": "mock-key"}}, limits={}, pricing={}, )