From f0c0741530f9ec06490707132c87f5ca1e224350 Mon Sep 17 00:00:00 2001 From: guenhter Date: Tue, 21 Apr 2026 07:28:32 +0200 Subject: [PATCH] feat: custom answers --- CHANGELOG.md | 6 + README.md | 33 ++- config.yaml | 8 + docs/llmock-skill/references/config.yaml | 8 + src/llmock/strategies/factory.py | 8 + src/llmock/strategies/strategy_composition.py | 7 +- .../strategies/strategy_custom_answers.py | 81 ++++++ tests/test_custom_answers_strategy.py | 236 ++++++++++++++++++ 8 files changed, 385 insertions(+), 2 deletions(-) create mode 100644 src/llmock/strategies/strategy_custom_answers.py create mode 100644 tests/test_custom_answers_strategy.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 7e99102..e6ac90b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.0.4] + +### Added + +- `CustomAnswersStrategy`: new strategy that returns preconfigured answers for exact-match questions. + ## [0.0.3] ### Added diff --git a/README.md b/README.md index 22e0480..f69f3d4 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,7 @@ OpenAI-compatible mock server for testing LLM integrations. - Default mirror strategy (echoes input as output) - **Tool calling support** — trigger phrase–driven tool call responses when `tools` are present in the request using `call tool '' with ''` - **Error simulation** — trigger phrase–driven error responses using `raise error ` in the last user message +- **Custom answers** — return preconfigured answers for exact-match questions via `customReplies` in `config.yaml` - Streaming support for both Chat Completions and Responses APIs (including `stream_options.include_usage`) ## Quick Start @@ -99,12 +100,20 @@ cors: - "http://localhost:8000" # Ordered list of strategies to try (first non-empty result wins) -# Available: ErrorStrategy, ToolCallStrategy, MirrorStrategy +# Available: ErrorStrategy, CustomAnswersStrategy, ToolCallStrategy, MirrorStrategy strategies: - ErrorStrategy + - CustomAnswersStrategy - ToolCallStrategy - MirrorStrategy +# Inline question→answer pairs for CustomAnswersStrategy (exact-match, case-sensitive) +customReplies: + - question: How are you today? + answer: I'm fine. how are you? + - question: What is 1+1? + answer: 2 + models: - id: "gpt-4o" created: 1715367049 @@ -218,6 +227,28 @@ client.chat.completions.create( Only the last user message is checked. System/assistant/tool messages are ignored. Works on both `/chat/completions` and `/responses` endpoints. +### Custom Answers + +When `CustomAnswersStrategy` is included in the `strategies` list, llmock checks the last user message against a list of preconfigured question/answer pairs (exact match, case-sensitive, whitespace trimmed). Configure the pairs inline in `config.yaml`: + +```yaml +strategies: + - ErrorStrategy + - CustomAnswersStrategy + - ToolCallStrategy + - MirrorStrategy + +customReplies: + - question: How are you today? + answer: I'm fine. how are you? + - question: What is 1+1? + answer: 2 +``` + +If the message matches a `question`, the corresponding `answer` is returned. If no match is found, the strategy falls through to the next one (e.g. `MirrorStrategy`). + +Works on both `/chat/completions` and `/responses` endpoints. + ## Development ### Run Tests diff --git a/config.yaml b/config.yaml index 5e51a70..747b08e 100644 --- a/config.yaml +++ b/config.yaml @@ -30,5 +30,13 @@ models: # Ordered list of strategies (first non-empty result wins) strategies: - ErrorStrategy + - CustomAnswersStrategy - ToolCallStrategy - MirrorStrategy + +# Inline question→answer pairs for CustomAnswersStrategy (exact-match, case-sensitive) +# customReplies: +# - question: How are you today? +# answer: I'm fine. how are you? +# - question: What is 1+1? +# answer: 2 diff --git a/docs/llmock-skill/references/config.yaml b/docs/llmock-skill/references/config.yaml index 5e51a70..747b08e 100644 --- a/docs/llmock-skill/references/config.yaml +++ b/docs/llmock-skill/references/config.yaml @@ -30,5 +30,13 @@ models: # Ordered list of strategies (first non-empty result wins) strategies: - ErrorStrategy + - CustomAnswersStrategy - ToolCallStrategy - MirrorStrategy + +# Inline question→answer pairs for CustomAnswersStrategy (exact-match, case-sensitive) +# customReplies: +# - question: How are you today? +# answer: I'm fine. how are you? +# - question: What is 1+1? +# answer: 2 diff --git a/src/llmock/strategies/factory.py b/src/llmock/strategies/factory.py index 312aab1..27f3f00 100644 --- a/src/llmock/strategies/factory.py +++ b/src/llmock/strategies/factory.py @@ -16,6 +16,10 @@ ChatErrorStrategy, ResponseErrorStrategy, ) +from llmock.strategies.strategy_custom_answers import ( + ChatCustomAnswersStrategy, + ResponseCustomAnswersStrategy, +) from llmock.strategies.strategy_tool_call import ( ChatToolCallStrategy, ResponseToolCallStrategy, @@ -23,6 +27,10 @@ # Short strategy names used in config → (ChatClass, ResponseClass) _STRATEGIES: dict[str, tuple[type, type]] = { + "CustomAnswersStrategy": ( + ChatCustomAnswersStrategy, + ResponseCustomAnswersStrategy, + ), "ErrorStrategy": (ChatErrorStrategy, ResponseErrorStrategy), "MirrorStrategy": (ChatMirrorStrategy, ResponseMirrorStrategy), "ToolCallStrategy": (ChatToolCallStrategy, ResponseToolCallStrategy), diff --git a/src/llmock/strategies/strategy_composition.py b/src/llmock/strategies/strategy_composition.py index e36a612..038eefb 100644 --- a/src/llmock/strategies/strategy_composition.py +++ b/src/llmock/strategies/strategy_composition.py @@ -22,7 +22,12 @@ logger = logging.getLogger(__name__) -_DEFAULT_STRATEGIES = ["ErrorStrategy", "ToolCallStrategy", "MirrorStrategy"] +_DEFAULT_STRATEGIES = [ + "ErrorStrategy", + "CustomAnswersStrategy", + "ToolCallStrategy", + "MirrorStrategy", +] class ChatCompositionStrategy: diff --git a/src/llmock/strategies/strategy_custom_answers.py b/src/llmock/strategies/strategy_custom_answers.py new file mode 100644 index 0000000..943fbb7 --- /dev/null +++ b/src/llmock/strategies/strategy_custom_answers.py @@ -0,0 +1,81 @@ +"""Custom answers strategy - return preconfigured answers for exact-match questions. + +Reads the ``customReplies`` list directly from config. The list must have +the following structure in ``config.yaml``:: + + customReplies: + - question: How are you today? + answer: I'm fine. how are you? + - question: What is 1+1? + answer: 2 + +When the last user message exactly matches a ``question`` (case-sensitive, +stripped of leading/trailing whitespace), the corresponding ``answer`` is +returned as a text response. If no match is found, an empty list is +returned so the next strategy in the composition chain runs. + +If ``customReplies`` is not present in config the strategy passes through +silently (empty list). +""" + +import logging +from typing import Any + +from llmock.schemas.chat import ChatCompletionRequest +from llmock.schemas.responses import ResponseCreateRequest +from llmock.strategies.base import StrategyResponse, text_response +from llmock.utils.chat import ( + extract_last_user_text_chat, + extract_last_user_text_response, +) + +logger = logging.getLogger(__name__) + +CONFIG_KEY = "customReplies" + + +def _build_replies(entries: list[Any]) -> dict[str, str]: + """Build a question→answer mapping from a list of config entries.""" + return { + str(entry["question"]).strip(): str(entry["answer"]) + for entry in entries + if isinstance(entry, dict) and "question" in entry and "answer" in entry + } + + +def _match(content: str | None, replies: dict[str, str]) -> list[StrategyResponse]: + """Return a response if *content* exactly matches a question, else [].""" + if content is None: + return [] + answer = replies.get(content.strip()) + if answer is None: + return [] + return [text_response(answer)] + + +class ChatCustomAnswersStrategy: + """Exact-match custom answers strategy for the Chat Completions API.""" + + def __init__(self, config: dict[str, Any]) -> None: + entries = config.get(CONFIG_KEY, []) + self._replies: dict[str, str] = _build_replies(entries) + + def generate_response( + self, request: ChatCompletionRequest + ) -> list[StrategyResponse]: + content = extract_last_user_text_chat(request) + return _match(content, self._replies) + + +class ResponseCustomAnswersStrategy: + """Exact-match custom answers strategy for the Responses API.""" + + def __init__(self, config: dict[str, Any]) -> None: + entries = config.get(CONFIG_KEY, []) + self._replies: dict[str, str] = _build_replies(entries) + + def generate_response( + self, request: ResponseCreateRequest + ) -> list[StrategyResponse]: + content = extract_last_user_text_response(request) + return _match(content, self._replies) diff --git a/tests/test_custom_answers_strategy.py b/tests/test_custom_answers_strategy.py new file mode 100644 index 0000000..07e6c5b --- /dev/null +++ b/tests/test_custom_answers_strategy.py @@ -0,0 +1,236 @@ +"""Tests for the custom answers strategy.""" + +from collections.abc import AsyncGenerator + +import httpx +import pytest + +from llmock.app import create_app +from llmock.config import Config, get_config +from llmock.schemas.chat import ChatCompletionRequest, ChatMessageRequest +from llmock.schemas.responses import ResponseCreateRequest +from llmock.strategies.strategy_custom_answers import ( + ChatCustomAnswersStrategy, + ResponseCustomAnswersStrategy, + _build_replies, +) + +TEST_API_KEY = "test-api-key" + +SAMPLE_ENTRIES = [ + {"question": "How are you today?", "answer": "I'm fine. how are you?"}, + {"question": "What is 1+1?", "answer": "2"}, +] + + +# ============================================================================ +# Unit tests - _build_replies +# ============================================================================ + + +def test_build_replies_returns_mapping() -> None: + result = _build_replies(SAMPLE_ENTRIES) + assert result["How are you today?"] == "I'm fine. how are you?" + assert result["What is 1+1?"] == "2" + + +def test_build_replies_empty_list() -> None: + assert _build_replies([]) == {} + + +def test_build_replies_skips_incomplete_entries() -> None: + result = _build_replies([{"question": "Only question"}, {"answer": "Only answer"}]) + assert result == {} + + +# ============================================================================ +# Unit tests - ChatCustomAnswersStrategy +# ============================================================================ + + +def test_chat_strategy_exact_match() -> None: + strategy = ChatCustomAnswersStrategy({"customReplies": SAMPLE_ENTRIES}) + request = ChatCompletionRequest( + model="gpt-4", + messages=[ChatMessageRequest(role="user", content="How are you today?")], + ) + result = strategy.generate_response(request) + assert len(result) == 1 + assert result[0].content == "I'm fine. how are you?" + + +def test_chat_strategy_no_match_returns_empty() -> None: + strategy = ChatCustomAnswersStrategy({"customReplies": SAMPLE_ENTRIES}) + request = ChatCompletionRequest( + model="gpt-4", + messages=[ + ChatMessageRequest(role="user", content="Something completely different") + ], + ) + result = strategy.generate_response(request) + assert result == [] + + +def test_chat_strategy_case_sensitive() -> None: + strategy = ChatCustomAnswersStrategy({"customReplies": SAMPLE_ENTRIES}) + request = ChatCompletionRequest( + model="gpt-4", + messages=[ChatMessageRequest(role="user", content="how are you today?")], + ) + result = strategy.generate_response(request) + assert result == [] + + +def test_chat_strategy_strips_whitespace() -> None: + strategy = ChatCustomAnswersStrategy({"customReplies": SAMPLE_ENTRIES}) + request = ChatCompletionRequest( + model="gpt-4", + messages=[ChatMessageRequest(role="user", content=" How are you today? ")], + ) + result = strategy.generate_response(request) + assert len(result) == 1 + assert result[0].content == "I'm fine. how are you?" + + +def test_chat_strategy_no_config_key_returns_empty() -> None: + strategy = ChatCustomAnswersStrategy({}) + request = ChatCompletionRequest( + model="gpt-4", + messages=[ChatMessageRequest(role="user", content="How are you today?")], + ) + result = strategy.generate_response(request) + assert result == [] + + +# ============================================================================ +# Unit tests - ResponseCustomAnswersStrategy +# ============================================================================ + + +def test_response_strategy_exact_match() -> None: + strategy = ResponseCustomAnswersStrategy({"customReplies": SAMPLE_ENTRIES}) + request = ResponseCreateRequest(model="gpt-4", input="What is 1+1?") + result = strategy.generate_response(request) + assert len(result) == 1 + assert result[0].content == "2" + + +def test_response_strategy_no_match_returns_empty() -> None: + strategy = ResponseCustomAnswersStrategy({"customReplies": SAMPLE_ENTRIES}) + request = ResponseCreateRequest(model="gpt-4", input="Unknown question") + result = strategy.generate_response(request) + assert result == [] + + +# ============================================================================ +# Integration tests - Chat Completions endpoint +# ============================================================================ + + +@pytest.fixture +def chat_config() -> Config: + return { + "models": [{"id": "gpt-4", "created": 1700000000, "owned_by": "openai"}], + "api-key": TEST_API_KEY, + "strategies": ["CustomAnswersStrategy", "MirrorStrategy"], + "customReplies": SAMPLE_ENTRIES, + } + + +@pytest.fixture +async def chat_client(chat_config: Config) -> AsyncGenerator[httpx.AsyncClient, None]: + app = create_app(config=chat_config) + app.dependency_overrides[get_config] = lambda: chat_config + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient( + transport=transport, + base_url="http://testserver", + headers={"Authorization": f"Bearer {TEST_API_KEY}"}, + ) as http_client: + yield http_client + + +async def test_chat_endpoint_returns_custom_answer( + chat_client: httpx.AsyncClient, +) -> None: + response = await chat_client.post( + "/chat/completions", + json={ + "model": "gpt-4", + "messages": [{"role": "user", "content": "What is 1+1?"}], + }, + ) + assert response.status_code == 200 + data = response.json() + assert data["choices"][0]["message"]["content"] == "2" + + +async def test_chat_endpoint_falls_through_to_mirror( + chat_client: httpx.AsyncClient, +) -> None: + response = await chat_client.post( + "/chat/completions", + json={ + "model": "gpt-4", + "messages": [{"role": "user", "content": "No match here"}], + }, + ) + assert response.status_code == 200 + data = response.json() + assert data["choices"][0]["message"]["content"] == "No match here" + + +# ============================================================================ +# Integration tests - Responses endpoint +# ============================================================================ + + +@pytest.fixture +def responses_config() -> Config: + return { + "models": [{"id": "gpt-4", "created": 1700000000, "owned_by": "openai"}], + "api-key": TEST_API_KEY, + "strategies": ["CustomAnswersStrategy", "MirrorStrategy"], + "customReplies": SAMPLE_ENTRIES, + } + + +@pytest.fixture +async def responses_client( + responses_config: Config, +) -> AsyncGenerator[httpx.AsyncClient, None]: + app = create_app(config=responses_config) + app.dependency_overrides[get_config] = lambda: responses_config + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient( + transport=transport, + base_url="http://testserver", + headers={"Authorization": f"Bearer {TEST_API_KEY}"}, + ) as http_client: + yield http_client + + +async def test_responses_endpoint_returns_custom_answer( + responses_client: httpx.AsyncClient, +) -> None: + response = await responses_client.post( + "/responses", + json={"model": "gpt-4", "input": "How are you today?"}, + ) + assert response.status_code == 200 + data = response.json() + output_text = data["output"][0]["content"][0]["text"] + assert output_text == "I'm fine. how are you?" + + +async def test_responses_endpoint_falls_through_to_mirror( + responses_client: httpx.AsyncClient, +) -> None: + response = await responses_client.post( + "/responses", + json={"model": "gpt-4", "input": "No match here"}, + ) + assert response.status_code == 200 + data = response.json() + output_text = data["output"][0]["content"][0]["text"] + assert output_text == "No match here"