Store Support Bot in Python: Source Code

1import json23import httpx4import openai5import pytest67from store_assistant.assistant import FALLBACK_BUSY, FALLBACK_EMPTY, FALLBACK_ERROR, StoreAssistant8from store_assistant.memory import ConversationMemory9from store_assistant.tools import ToolExecutor1011from .conftest import FakeClient, completion, tool_call121314def make_assistant(store, client, *, history=12, rounds=4):15    return StoreAssistant(16        client=client,17        model="test-model",18        executor=ToolExecutor(store),19        memory=ConversationMemory(history),20        system_prompt="SYSTEM",21        max_tool_rounds=rounds,22        max_output_tokens=256,23    )242526async def test_plain_answer(store) -> None:27    client = FakeClient([completion("Salam! Kifach n3awnek?")])28    reply = await make_assistant(store, client).reply(1, "salam")29    assert reply.text == "Salam! Kifach n3awnek?"30    assert reply.tools_used == []31    request = client.requests[0]32    assert request["model"] == "test-model"33    assert request["tool_choice"] == "auto"34    assert request["messages"][0] == {"role": "system", "content": "SYSTEM"}353637async def test_tool_call_then_answer(store) -> None:38    client = FakeClient(39        [40            completion(tool_calls=[tool_call("c1", "get_delivery_fee", {"wilaya": "وهران"})]),41            completion("التوصيل لوهران: 650 دج للدار، 400 دج للمكتب."),42        ]43    )44    reply = await make_assistant(store, client).reply(1, "قداه التوصيل لوهران؟")45    assert reply.tools_used == ["get_delivery_fee"]46    assert "650" in reply.text4748    second = client.requests[1]["messages"]49    assistant_turn, tool_turn = second[-2], second[-1]50    assert assistant_turn["tool_calls"][0]["function"]["name"] == "get_delivery_fee"51    assert tool_turn["role"] == "tool" and tool_turn["tool_call_id"] == "c1"52    assert json.loads(tool_turn["content"])["home_delivery_dzd"] == 650535455async def test_parallel_tool_calls(store) -> None:56    client = FakeClient(57        [58            completion(59                tool_calls=[60                    tool_call("a", "get_delivery_fee", {"wilaya": "Alger"}),61                    tool_call("b", "search_products", {"query": "sneakers", "size": "42"}),62                ]63            ),64            completion("ok"),65        ]66    )67    reply = await make_assistant(store, client).reply(1, "sneakers 42 + livraison Alger ?")68    assert reply.tools_used == ["get_delivery_fee", "search_products"]69    tool_ids = [m["tool_call_id"] for m in client.requests[1]["messages"] if m["role"] == "tool"]70    assert tool_ids == ["a", "b"]717273async def test_tool_rounds_are_capped(store) -> None:74    looping = completion(tool_calls=[tool_call("x", "get_store_policies", {})])75    client = FakeClient([looping, looping, completion("final answer")])76    reply = await make_assistant(store, client, rounds=2).reply(1, "policies?")77    assert reply.text == "final answer"78    assert [r["tool_choice"] for r in client.requests] == ["auto", "auto", "none"]798081async def test_ticket_ids_are_returned(store) -> None:82    client = FakeClient(83        [84            completion(tool_calls=[tool_call("t", "request_human_agent", {"reason": "wants to cancel NS-10242"})]),85            completion("Un conseiller va vous répondre ici."),86        ]87    )88    reply = await make_assistant(store, client).reply(9, "je veux annuler", username="amina")89    assert len(reply.ticket_ids) == 1909192async def test_history_is_kept_and_capped(store) -> None:93    client = FakeClient([completion(f"answer {i}") for i in range(4)])94    assistant = make_assistant(store, client, history=4)95    for i in range(4):96        await assistant.reply(5, f"question {i}")97    last = client.requests[-1]["messages"]98    # system + 4 remembered messages (2 exchanges) + new question99    assert [m["content"] for m in last] == ["SYSTEM", "question 1", "answer 1", "question 2", "answer 2", "question 3"]100101102async def test_history_is_per_chat_and_resettable(store) -> None:103    client = FakeClient([completion("a"), completion("b"), completion("c")])104    assistant = make_assistant(store, client)105    await assistant.reply(1, "first chat")106    await assistant.reply(2, "second chat")107    assert len(client.requests[1]["messages"]) == 2108    assistant.reset(1)109    await assistant.reply(1, "again")110    assert len(client.requests[2]["messages"]) == 2111112113async def test_empty_answer_falls_back_and_is_not_remembered(store) -> None:114    client = FakeClient([completion(""), completion("ok")])115    assistant = make_assistant(store, client)116    assert (await assistant.reply(1, "hmm")).text == FALLBACK_EMPTY117    await assistant.reply(1, "hello")118    assert len(client.requests[1]["messages"]) == 2119120121def _request() -> httpx.Request:122    return httpx.Request("POST", "https://api.example.test/v1/chat/completions")123124125@pytest.mark.parametrize(126    ("error", "expected"),127    [128        (openai.APIConnectionError(request=_request()), FALLBACK_ERROR),129        (openai.APITimeoutError(request=_request()), FALLBACK_ERROR),130    ],131)132async def test_network_errors_return_fallback(store, error, expected) -> None:133    reply = await make_assistant(store, FakeClient([error])).reply(1, "salam")134    assert reply.failed is True135    assert reply.text == expected136137138async def test_rate_limit_and_status_errors(store) -> None:139    rate_limited = openai.RateLimitError("slow down", response=httpx.Response(429, request=_request()), body=None)140    server_error = openai.InternalServerError("boom", response=httpx.Response(500, request=_request()), body=None)141    assistant = make_assistant(store, FakeClient([rate_limited, server_error]))142    assert (await assistant.reply(1, "a")).text == FALLBACK_BUSY143    assert (await assistant.reply(1, "b")).text == FALLBACK_ERROR144145146async def test_repeated_tool_calls_run_once(store) -> None:147    client = FakeClient(148        [149            completion(150                tool_calls=[151                    tool_call("a", "get_store_policies", {}),152                    tool_call("b", "get_store_policies", {}),153                    tool_call("c", "request_human_agent", {"reason": "exchange NS-10242"}),154                ]155            ),156            completion(tool_calls=[tool_call("d", "request_human_agent", {"reason": "exchange NS-10242"})]),157            completion("ok"),158        ]159    )160    reply = await make_assistant(store, client).reply(1, "exchange please")161    assert reply.tools_used == ["get_store_policies", "request_human_agent"]162    assert len(reply.ticket_ids) == 1163    # every call id still gets its own tool message164    tool_ids = [m["tool_call_id"] for m in client.requests[2]["messages"] if m["role"] == "tool"]165    assert tool_ids == ["a", "b", "c", "d"]166167168async def test_language_instruction_is_sent_before_the_message_but_not_remembered(store) -> None:169    client = FakeClient([completion("Yes."), completion("ok")])170    assistant = make_assistant(store, client)171    await assistant.reply(1, "Can you deliver to Oran?")172    first = client.requests[0]["messages"]173    assert first[-2]["role"] == "system" and "English" in first[-2]["content"]174    assert first[-1] == {"role": "user", "content": "Can you deliver to Oran?"}175    await assistant.reply(1, "4821")176    second = client.requests[1]["messages"]177    # history keeps only user/assistant text, and "4821" gets no language instruction178    assert [m["role"] for m in second] == ["system", "user", "assistant", "user"]