Créer un bot de support IA pour votre boutique en ligne

Bot de support en Python : code source

Fichiers : 2988,4 KiBTélécharger le ZIP

store_assistant/__main__.py

5,2 KiBBrutTélécharger
Python
1"""Command line entry point.23python -m store_assistant init-db [--reset]   create the database from the catalog file4python -m store_assistant chat                talk to the assistant in the terminal5python -m store_assistant bot                 run the Telegram bot6"""78from __future__ import annotations910import argparse11import asyncio12import logging13import sys1415from openai import AsyncOpenAI1617from .assistant import StoreAssistant18from .config import ConfigError, Settings19from .memory import ConversationMemory20from .prompts import build_system_prompt21from .store import Store, open_store22from .tools import ToolExecutor2324CLI_CHAT_ID = 0252627def build_assistant(settings: Settings, store: Store) -> tuple[StoreAssistant, str]:28    store_name = store.store_info().get("name", "")29    client = AsyncOpenAI(30        api_key=settings.devup_api_key,31        base_url=settings.devup_base_url,32        timeout=settings.request_timeout_s,33        max_retries=2,34    )35    assistant = StoreAssistant(36        client=client,37        model=settings.model,38        executor=ToolExecutor(store),39        memory=ConversationMemory(settings.history_messages),40        system_prompt=build_system_prompt(store_name),41        max_tool_rounds=settings.max_tool_rounds,42        max_output_tokens=settings.max_output_tokens,43    )44    return assistant, store_name454647def cmd_init_db(settings: Settings, reset: bool) -> int:48    settings.db_path.parent.mkdir(parents=True, exist_ok=True)49    store = Store(settings.db_path)50    try:51        if reset or store.is_empty():52            store.load_catalog(settings.catalog_path, replace=reset)53            print(f"Database ready: {settings.db_path}")54        else:55            print(f"Database already has data: {settings.db_path} (use --reset to reload the catalog)")56    finally:57        store.close()58    return 0596061async def _chat_loop(assistant: StoreAssistant, store_name: str, show_tools: bool) -> None:62    print(f"{store_name or 'Store'} assistant. Type /reset for a new conversation, /quit to exit.\n")63    while True:64        try:65            text = (await asyncio.to_thread(input, "you> ")).strip()66        except (EOFError, KeyboardInterrupt):67            print()68            return69        if not text:70            continue71        if text in {"/quit", "/exit"}:72            return73        if text == "/reset":74            assistant.reset(CLI_CHAT_ID)75            print("(new conversation)\n")76            continue77        reply = await assistant.reply(CLI_CHAT_ID, text, username="terminal")78        if show_tools and reply.tools_used:79            print(f"  [tools: {', '.join(reply.tools_used)}]")80        print(f"bot> {reply.text}\n")818283def cmd_chat(settings: Settings, show_tools: bool) -> int:84    store = open_store(settings.db_path, settings.catalog_path)85    try:86        assistant, store_name = build_assistant(settings, store)87        asyncio.run(_chat_loop(assistant, store_name, show_tools))88    finally:89        store.close()90    return 0919293def cmd_bot(settings: Settings) -> int:94    from .bot import SupportBot  # imported here so the other commands work without Telegram setup9596    store = open_store(settings.db_path, settings.catalog_path)97    try:98        assistant, store_name = build_assistant(settings, store)99        app = SupportBot(settings, assistant, store_name).build_application()100        logging.getLogger(__name__).info("bot started with model %s", settings.model)101        app.run_polling(allowed_updates=["message"], drop_pending_updates=True)102    finally:103        store.close()104    return 0105106107def main(argv: list[str] | None = None) -> int:108    parser = argparse.ArgumentParser(prog="store_assistant", description="AI customer support assistant")109    parser.add_argument("-v", "--verbose", action="store_true", help="debug logging")110    sub = parser.add_subparsers(dest="command", required=True)111    init_db = sub.add_parser("init-db", help="create the database from the catalog file")112    init_db.add_argument("--reset", action="store_true", help="delete existing data and reload the catalog")113    chat = sub.add_parser("chat", help="talk to the assistant in the terminal")114    chat.add_argument("--show-tools", action="store_true", help="print which tools each reply used")115    sub.add_parser("bot", help="run the Telegram bot")116    args = parser.parse_args(argv)117118    logging.basicConfig(119        level=logging.DEBUG if args.verbose else logging.INFO,120        format="%(asctime)s %(levelname)s %(name)s: %(message)s",121    )122    # The HTTP clients log every request URL at INFO, and Telegram URLs contain the bot token.123    for noisy in ("httpx", "httpx2"):124        logging.getLogger(noisy).setLevel(logging.WARNING)125    if args.command == "chat" and not args.verbose:126        logging.getLogger().setLevel(logging.WARNING)127128    try:129        if args.command == "init-db":130            return cmd_init_db(Settings.from_env(require_api_key=False), args.reset)131        settings = Settings.from_env(require_telegram=args.command == "bot")132    except ConfigError as exc:133        print(f"Configuration error: {exc}", file=sys.stderr)134        return 2135136    if args.command == "chat":137        return cmd_chat(settings, args.show_tools)138    return cmd_bot(settings)139140141if __name__ == "__main__":142    sys.exit(main())