diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md new file mode 100644 index 0000000..f4e7d33 --- /dev/null +++ b/.github/copilot-instructions.md @@ -0,0 +1,52 @@ +## Quick Agent Instructions for Houshan API + +This repository is a FastAPI service with Tortoise ORM (Aerich) migrations, several AI integrations, and Docker-based developer workflows. Use these notes to be productive quickly. + +- **Big picture**: API server in [src/main.py](src/main.py) mounts routes from `src/routes` and models from `src/models`. DB uses Tortoise ORM configured in [src/settings.py](src/settings.py) and [src/config.py](src/config.py). Background jobs use APScheduler (timezone Asia/Tehran). Key AI integrations include Fal (`FAL_KEY`), Suno (`SUNO_API_KEY`), OpenAI/Anthropic, and Chroma (see `requirements.txt`). + +- **Run / dev workflows**: + - Local (fast): run with Uvicorn: `uvicorn src.main:app --reload --host 0.0.0.0 --port 8000` (requires env from `.api.env`). + - Docker: use the Makefile targets which wrap `docker compose`: + - `make up` — build & start services + - `make logs` — follow logs + - `make shell-app` — open a shell inside the `app` container + - `docker-compose.yml` uses env file `.api.env` and exposes the API on port 8000 inside the container. See [docker-compose.yml](docker-compose.yml). + +- **Environment & config**: + - Environment variables live in `.api.env` (referenced by `Settings.model_config` in [src/config.py](src/config.py)). Always copy production secrets out of commits; `serviceAccountKey.json` is present and sensitive. + - DB URL lives in `DATABASE_URL` and is referenced from `src/settings.py` and `pyproject.toml` via Aerich config. + +- **Migrations**: + - Aerich is configured in `pyproject.toml` (`tool.aerich.tortoise_orm = "settings.TORTOISE_ORM"`). Typical commands (run with proper `DATABASE_URL` in env): + - `aerich init -t src.settings.TORTOISE_ORM` + - `aerich migrate --name "describe"` + - `aerich upgrade` + - Migrations live under `migrations/models`. + +- **Code layout & conventions**: + - `src/routes` – routers are registered centrally by `register_routers(app)` in [src/main.py](src/main.py). + - `src/services` – contains pluggable service modules (storage, tool, chatbot, etc.). Prefer calling service functions rather than re-implementing integrations. + - `src/tasks` – background jobs and scheduled tasks (scheduled from `src/main.py` via APScheduler). + - Use async/await consistently; many IO layers (httpx, aioboto3, asyncpg, tortoise) are async-first. + +- **Integration tips & patterns (project-specific)**: + - Fal video generation: endpoint implemented in [src/main.py](src/main.py) under `/public/generate-video`. It expects `FAL_KEY` and uses `FAL_BASE_URL`. Keep the key in `.api.env`. + - Suno: there is a webhook `/suno/callback` that writes to `SunoTask` (see `src/main.py`). When testing webhooks locally, route requests to the container or use `ngrok`. + - Storage: `src/services/storage.py` abstracts file storage (S3/compatible). Use it for large files instead of returning raw streams. + - Database model lookups use `await Model.get_or_none(...)` — follow that idiom to avoid exceptions. + +- **Testing & debugging**: + - No automated test harness found. Use `uvicorn` locally or `make up` + `make logs` for smoke tests. + - For DB schema issues, run `aerich migrate`/`upgrade` and inspect `migrations/models`. + - To inspect runtime logs or DB state in containers: `make logs` and `make shell-app`. + +- **Where to change things (examples)**: + - Add a new API router: create file in `src/routes`, export a `router`, and ensure `register_routers` imports it. + - Add a scheduled job: add function in `src/tasks/*` and register it in `src/main.py` with `scheduler.add_job(..., CronTrigger(..., timezone=tehran_tz))`. + +- **Caveats & gotchas**: + - The project is async-first; avoid blocking calls (do CPU-heavy work in separate workers). + - Several keys and external services (Fal, Suno, Chroma, Google, MailerSend) are used—stub or mock them when necessary. + - Aerich migrations may be sensitive to `TORTOISE_ORM` path; keep `pyproject.toml` and `src/settings.py` in sync. + +If any section is unclear or you'd like more examples (e.g., how to add a route or run aerich locally on Windows), tell me which part to expand. diff --git a/src/services/chatbot.py b/src/services/chatbot.py index 719a803..b77dab2 100644 --- a/src/services/chatbot.py +++ b/src/services/chatbot.py @@ -31,7 +31,7 @@ from langchain_core.messages import ( ) from .database.service import DatabaseService -from .kiaai import get_kia_model +from .kiaai import get_kia_model, kia_stream_and_save from ..config import settings from ..models import Chatbot, Message, BotStatus, BotType from ..messages import MESSAGES @@ -173,11 +173,7 @@ def audio_loader(content, query: str, audio: [str | None, str | None]): return content def chatbot_workflow(bot, now): - is_kia_model = bot.model.startswith("kia/") - print(f"[chatbot_workflow] START | model='{bot.model}' | is_kia={is_kia_model} | web_search={bot.web_search}") - model = get_model(model=bot.model) - print(f"[chatbot_workflow] Model initialized: {type(model).__name__ if model else 'KiaAIService (direct http)'}") trimmer = get_trimmer() if bot.web_search and model is not None: @@ -185,13 +181,9 @@ def chatbot_workflow(bot, now): async def call_model(state: MessagesState): kwargs = {} - is_kia = bot.model.startswith("kia/") - if "gemini" not in bot.model and not is_kia: + if "gemini" not in bot.model: kwargs["stream_usage"] = True - if is_kia: - print(f"[KiaAI] call_model invoked | model='{bot.model}' | stream_usage disabled") - trimmed_messages = await trimmer.ainvoke(state["messages"]) input_messages = [ SystemMessage( @@ -204,28 +196,7 @@ def chatbot_workflow(bot, now): ) ] + trimmed_messages - # KIA models must bypass LangChain entirely: - # both ainvoke() and invoke() internally use _astream()/_stream() in this - # version of LangChain which fails on KIA's choices=null streaming chunks. - # Direct httpx call is the only reliable path. - if is_kia: - from .kiaai import KiaAIService - print(f"[KiaAI] Calling via KiaAIService.invoke() | model='{bot.model}' | web_search={bot.web_search}") - response = await KiaAIService().invoke( - model=bot.model, - messages=input_messages, - google_search=bool(bot.web_search), - ) - else: - response = await model.ainvoke(input=input_messages, **kwargs) - - if is_kia: - print(f"[KiaAI] Response received | type={type(response).__name__}") - print(f"[KiaAI] content='{response.content[:200] if response.content else None}'") - print(f"[KiaAI] usage_metadata={response.usage_metadata}") - print(f"[KiaAI] additional_kwargs={response.additional_kwargs}") - print(f"[KiaAI] tool_calls={response.tool_calls}") - + response = await model.ainvoke(input=input_messages, **kwargs) ai_msg_created_at = datetime.now(UTC).isoformat() response.additional_kwargs["created_at"] = ai_msg_created_at @@ -451,63 +422,22 @@ async def send_message( human_msg.additional_kwargs["created_at"] = human_msg_created_at input_messages = [human_msg] - # ── KIA path: stream tokens directly, bypass LangGraph astream ────────── - # LangGraph delivers an AIMessage as ONE chunk (not token-by-token) because - # call_model returns a complete AIMessage. For real streaming we drive - # KiaAIService.stream() here and manage the graph state manually. + # ── KIA path: token-by-token streaming via kiaai.workflow ────────────── if bot.model.startswith("kia/"): - from .kiaai import KiaAIService - - # Push the human message into the graph state first - await app.aupdate_state(config, {"messages": [human_msg]}, as_node="model") - state = await app.aget_state(config) - all_msgs = state.values.get("messages", [human_msg]) - trimmed = await get_trimmer().ainvoke(all_msgs) - - kia_input = [ - SystemMessage( - f"Today: {now.isoformat()}." - "You are an AI assistant designed to answer user queries conversationally, using tools when necessary." - "Follow these instructions carefully: " - "1. **Mathematical Formatting**: When responding to questions involving mathematics, format all mathematical expressions and formulas using Markdown with LaTeX notation for clarity and readability. Use inline LaTeX (e.g., `$x^2 + 2x + 1$`) for expressions within a sentence, and display LaTeX (e.g., `$$x^2 + 2x + 1 = 0$$`) for standalone equations or complex formulas. Ensure the formatting is compatible with Markdown renderers that support LaTeX." - f"{bot.prompt if bot.web_search else ''}" - "Follow these guidelines to ensure a seamless and informative conversation with the user." - ) - ] + trimmed - - print(f"[KiaAI] Direct stream | model='{bot.model}' | web_search={bot.web_search} | history_len={len(trimmed)}") - - full_ai_content = "" - ai_msg_created_at = datetime.now(UTC).isoformat() - try: - async for chunk in KiaAIService().stream( - model=bot.model, - messages=kia_input, - google_search=bool(bot.web_search), - ): - token = chunk.content or "" - if token: - full_ai_content += token - yield json.dumps({"content": token}, ensure_ascii=False) - if chunk.usage_metadata: - input_tokens = chunk.usage_metadata.get("input_tokens", 0) - output_tokens = chunk.usage_metadata.get("output_tokens", 0) + async for item in kia_stream_and_save(app, config, human_msg, bot, now, get_trimmer()): + if isinstance(item, dict): + input_tokens = item["input_tokens"] + output_tokens = item["output_tokens"] + ai_msg_created_at = item["ai_msg_created_at"] + human_msg_id, ai_msg_id = get_message_id(item["final_msgs"]) + else: + yield item except Exception as e: print(f"[KiaAI] Streaming error: {e}") yield json.dumps({"error": True, "detail": str(e)}, ensure_ascii=False) return - # Save AI response back into the graph state - ai_msg = AIMessage(content=full_ai_content) - ai_msg.usage_metadata = { - "input_tokens": input_tokens, - "output_tokens": output_tokens, - "total_tokens": input_tokens + output_tokens, - } - ai_msg.additional_kwargs["created_at"] = ai_msg_created_at - await app.aupdate_state(config, {"messages": [ai_msg]}, as_node="model") - await user.save() await Message.create( input_tokens=input_tokens, @@ -518,11 +448,6 @@ async def send_message( cost_from_credit=usage_report["credit"], cost_from_free=usage_report["free"], ) - - final_state = await app.aget_state(config) - final_msgs = final_state.values["messages"] - human_msg_id, ai_msg_id = get_message_id(final_msgs) - yield json.dumps({ "credit": user.credit, "free": user.free_credit, diff --git a/src/services/kiaai/__init__.py b/src/services/kiaai/__init__.py index 5f3b141..f5bf000 100644 --- a/src/services/kiaai/__init__.py +++ b/src/services/kiaai/__init__.py @@ -1,3 +1,4 @@ from .chat import KiaAIService, get_kia_model +from .workflow import kia_stream_and_save -__all__ = ["KiaAIService", "get_kia_model"] +__all__ = ["KiaAIService", "get_kia_model", "kia_stream_and_save"] diff --git a/src/services/kiaai/workflow.py b/src/services/kiaai/workflow.py new file mode 100644 index 0000000..67f1a36 --- /dev/null +++ b/src/services/kiaai/workflow.py @@ -0,0 +1,97 @@ +import json +from datetime import datetime, UTC +from typing import AsyncGenerator + +from langchain_core.messages import AIMessage, HumanMessage, SystemMessage + +from .chat import KiaAIService + + +def _build_system_message(now, bot) -> SystemMessage: + return SystemMessage( + f"Today: {now.isoformat()}. " + "You are an AI assistant designed to answer user queries conversationally, using tools when necessary. " + "Follow these instructions carefully: " + "1. **Mathematical Formatting**: When responding to questions involving mathematics, format all mathematical " + "expressions and formulas using Markdown with LaTeX notation for clarity and readability. " + "Use inline LaTeX (e.g., `$x^2 + 2x + 1$`) for expressions within a sentence, and display LaTeX " + "(e.g., `$$x^2 + 2x + 1 = 0$$`) for standalone equations or complex formulas. " + "Ensure the formatting is compatible with Markdown renderers that support LaTeX. " + f"{bot.prompt if bot.web_search else ''}" + "Follow these guidelines to ensure a seamless and informative conversation with the user." + ) + + +async def kia_stream_and_save( + app, + config: dict, + human_msg: HumanMessage, + bot, + now, + trimmer, +) -> AsyncGenerator: + """ + KIA direct-stream workflow. Drives KiaAIService.stream() and manages + LangGraph state manually (bypassing app.astream entirely). + + Yields: + str – JSON-encoded token chunks: ``{"content": "..."}`` + dict – final metadata (always the last yielded item), keys: + ``input_tokens``, ``output_tokens``, ``ai_msg_created_at``, + ``final_msgs`` + + Raises: + Any exception from KiaAIService.stream() is propagated to the caller. + If an exception is raised after tokens have already been yielded, the + graph state will contain the human message but no AI reply. + """ + # 1. Push human message into graph state + await app.aupdate_state(config, {"messages": [human_msg]}, as_node="model") + state = await app.aget_state(config) + all_msgs = state.values.get("messages", [human_msg]) + trimmed = await trimmer.ainvoke(all_msgs) + + kia_input = [_build_system_message(now, bot)] + trimmed + + print( + f"[KiaAI] Direct stream | model='{bot.model}' " + f"| web_search={bot.web_search} | history_len={len(trimmed)}" + ) + + full_ai_content = "" + ai_msg_created_at = datetime.now(UTC).isoformat() + input_tokens = 0 + output_tokens = 0 + + # 2. Stream tokens — exceptions propagate to the caller + async for chunk in KiaAIService().stream( + model=bot.model, + messages=kia_input, + google_search=bool(bot.web_search), + ): + token = chunk.content or "" + if token: + full_ai_content += token + yield json.dumps({"content": token}, ensure_ascii=False) + if chunk.usage_metadata: + input_tokens = chunk.usage_metadata.get("input_tokens", 0) + output_tokens = chunk.usage_metadata.get("output_tokens", 0) + + # 3. Save completed AI message back into graph state + ai_msg = AIMessage(content=full_ai_content) + ai_msg.usage_metadata = { + "input_tokens": input_tokens, + "output_tokens": output_tokens, + "total_tokens": input_tokens + output_tokens, + } + ai_msg.additional_kwargs["created_at"] = ai_msg_created_at + await app.aupdate_state(config, {"messages": [ai_msg]}, as_node="model") + + # 4. Yield final metadata for the caller to persist and respond with + final_state = await app.aget_state(config) + yield { + "input_tokens": input_tokens, + "output_tokens": output_tokens, + "ai_msg_created_at": ai_msg_created_at, + "final_msgs": final_state.values["messages"], + }