playchat/backend/app/core/llm.py

import json
import logging
import time
import traceback
from typing import TYPE_CHECKING, AsyncIterator, AsyncGenerator
from contextlib import asynccontextmanager

import litellm
from app.core.config import DEFAULT_MODEL, DEBUG
from app.core.notes import NOTE_TOOLS, execute_note_tool
from app.core.search.websearch import WEB_SEARCH_TOOL, search_web

if TYPE_CHECKING:
    from sqlalchemy.orm import Session as DBSession

litellm.drop_params = True
if DEBUG:
    litellm._turn_on_debug()

logger = logging.getLogger(__name__)


@asynccontextmanager
async def _log_context(
    model: str,
    messages: list[dict],
    session_id: str | None = None,
) -> AsyncGenerator[dict, None]:
    """LLM呼び出しの前後でログデータを収集するコンテキストマネージャー。"""
    log: dict = {
        "session_id": session_id,
        "model": model,
        "request_json": json.dumps(messages, ensure_ascii=False),
        "response_text": None,
        "error": None,
        "latency_ms": None,
    }
    start = time.perf_counter()
    try:
        yield log
    except Exception as e:
        log["error"] = str(e)
        log["latency_ms"] = round((time.perf_counter() - start) * 1000, 1)
        logger.error("LLM call failed [model=%s]: %s\n%s", log["model"], e, traceback.format_exc())
        raise
    finally:
        if log["latency_ms"] is None:
            log["latency_ms"] = round((time.perf_counter() - start) * 1000, 1)


def _save_log(log: dict) -> None:
    from app.db.database import SessionLocal
    from app.db.models import ApiLog
    db = SessionLocal()
    try:
        db.add(ApiLog(**log))
        db.commit()
    finally:
        db.close()


async def stream_chat(
    messages: list[dict],
    model: str | None = None,
    session_id: str | None = None,
    tools: list[dict] | None = None,
) -> AsyncIterator[str]:
    used_model = model or DEFAULT_MODEL
    async with _log_context(used_model, messages, session_id) as log:
        response = await litellm.acompletion(
            model=used_model,
            messages=messages,
            stream=True,
            **({"tools": tools} if tools else {}),
        )
        chunks: list[str] = []
        async for chunk in response:
            delta = chunk.choices[0].delta.content
            if delta:
                chunks.append(delta)
                yield delta
        log["response_text"] = "".join(chunks)
    _save_log(log)


async def complete(
    messages: list[dict],
    model: str | None = None,
    session_id: str | None = None,
) -> str:
    used_model = model or DEFAULT_MODEL
    async with _log_context(used_model, messages, session_id) as log:
        response = await litellm.acompletion(
            model=used_model,
            messages=messages,
            stream=False,
        )
        result = response.choices[0].message.content or ""
        log["response_text"] = result
    _save_log(log)
    return result


async def embed(texts: list[str], model: str) -> list[list[float]]:
    """テキスト群を埋め込みベクトルに変換する。"""
    response = await litellm.aembedding(model=model, input=texts)
    return [item["embedding"] for item in response.data]


async def resolve_tool_calls(
    messages: list[dict],
    model: str | None = None,
    session_id: str | None = None,
    db: "DBSession | None" = None,
    max_rounds: int = 2,
) -> tuple[list[dict], list[dict], bool]:
    """必要に応じてLLMにツールを呼ばせ、その結果を会話に組み込んだ上でメッセージ列を返す。

    扱うツールは web_search と共有ノート（read_note / write_note）。ノート系は db が
    渡されたときだけ提示する。モデルがtool callingに対応していない場合は何もせず
    元のmessagesをそのまま返す。
    戻り値: (拡張後のmessages, 実際に検索したクエリと結果のリスト, ノートを変更したか)
    """
    used_model = model or DEFAULT_MODEL
    working_messages = list(messages)
    searches: list[dict] = []
    notes_changed = False

    tools = [WEB_SEARCH_TOOL] + (NOTE_TOOLS if db is not None else [])

    for _ in range(max_rounds):
        try:
            response = await litellm.acompletion(
                model=used_model,
                messages=working_messages,
                tools=tools,
                tool_choice="auto",
            )
        except Exception:
            logger.info("Model %s does not support tool calling; skipping tools", used_model)
            return messages, searches, notes_changed

        choice_message = response.choices[0].message
        tool_calls = getattr(choice_message, "tool_calls", None)
        if not tool_calls:
            break

        working_messages.append({
            "role": "assistant",
            "content": choice_message.content or "",
            "tool_calls": [
                {
                    "id": tc.id,
                    "type": "function",
                    "function": {"name": tc.function.name, "arguments": tc.function.arguments},
                }
                for tc in tool_calls
            ],
        })

        for tc in tool_calls:
            try:
                args = json.loads(tc.function.arguments or "{}")
            except json.JSONDecodeError:
                args = {}

            name = tc.function.name
            if name in ("read_note", "write_note") and db is not None:
                tool_text, changed = execute_note_tool(db, name, args)
                notes_changed = notes_changed or changed
            else:
                query = args.get("query", "")
                results = await search_web(query)
                if results:
                    tool_text = "\n\n".join(f"[{r['title']}]({r['url']})\n{r['content']}" for r in results)
                    searches.append({"query": query, "results": results})
                else:
                    tool_text = "検索結果が見つかりませんでした。"

            working_messages.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": tool_text,
            })

    return working_messages, searches, notes_changed