"""ユーザーとAIが共同で編集する共有ノート。 システムプロンプトにはページの「タイトル一覧」だけを注入し、本文はモデルが read_note ツールで必要なときだけ読み出す。書き込みは write_note ツールで行う。 """ import logging import re from sqlalchemy.orm import Session as DBSession from app.core.config import NOTES_TOOL_CONTENT_LIMIT from app.db.models import NotePage logger = logging.getLogger(__name__) READ_NOTE_TOOL = { "type": "function", "function": { "name": "read_note", "description": "共有ノートの特定ページの本文を読み出す。システムプロンプトに提示されたタイトル一覧の中から、" "今の会話に関係しそうなページがあるときに使う。", "parameters": { "type": "object", "properties": { "title": {"type": "string", "description": "読み出すページのタイトル（タイトル一覧に載っているもの）"}, }, "required": ["title"], }, }, } WRITE_NOTE_TOOL = { "type": "function", "function": { "name": "write_note", "description": "共有ノートのページに書き込む。ユーザーからノートへの記録・追記・整理を頼まれたときに使う。" "存在しないタイトルを指定した場合は新しいページを作成する。" "既存ページを replace で上書きすると元の内容は失われるので、追記でよい場合は必ず append を使うこと。", "parameters": { "type": "object", "properties": { "title": {"type": "string", "description": "書き込む対象のページタイトル"}, "content": {"type": "string", "description": "書き込む本文（Markdown可）"}, "mode": { "type": "string", "enum": ["append", "replace"], "description": "append=末尾に追記（既定） / replace=ページ全体を置き換え", }, }, "required": ["title", "content"], }, }, } NOTE_TOOLS = [READ_NOTE_TOOL, WRITE_NOTE_TOOL] def _find_page(db: DBSession, title: str) -> NotePage | None: """タイトルでページを探す。完全一致 → 大文字小文字・前後空白を無視した一致 の順に試す。""" pages = db.query(NotePage).order_by(NotePage.position, NotePage.id).all() for page in pages: if page.title == title: return page normalized = title.strip().casefold() for page in pages: if page.title.strip().casefold() == normalized: return page return None def next_position(db: DBSession) -> int: """新規ページを末尾に置くための position を返す。""" last = db.query(NotePage).order_by(NotePage.position.desc()).first() return (last.position + 1) if last else 0 def set_section(db: DBSession, page_id: int, section: str | None) -> NotePage | None: """ページの所属セクションを変更し、同じセクションのページ群の直後に連続配置し直す。 タブバーはposition順に描画され、隣接して同じsectionが続く区間を1つのグループとして まとめて表示する（app.api.notes / フロントのグルーピング表示を参照）。そのため セクションを変更したページは、常に同じセクションの最後のページの直後に移動する。 """ page = db.get(NotePage, page_id) if not page: return None normalized = (section or "").strip() or None ordered = [p for p in db.query(NotePage).order_by(NotePage.position, NotePage.id).all() if p.id != page_id] if normalized is None: insert_at = len(ordered) else: matching = [i for i, p in enumerate(ordered) if p.section == normalized] insert_at = (matching[-1] + 1) if matching else len(ordered) ordered.insert(insert_at, page) page.section = normalized for i, p in enumerate(ordered): p.position = i db.commit() db.refresh(page) return page def reorder(db: DBSession, page_id: int, section: str | None, order: list[int]) -> bool: """タブのドラッグ&ドロップによる並べ替えを反映する。 order にはドラッグ後の全ページIDを新しい順序で渡す（フロント側で計算する）。 page_id のページには先に section を設定し、その後 order の順に position を振り直す。 こうすることで、別セクションのタブの前後にドロップした場合はそのセクションへの 移動を兼ね、同じセクション内でのドロップなら単純な並べ替えになる。 order が現在の全ページIDの集合と一致しない場合は何もせず False を返す。 """ pages = {p.id: p for p in db.query(NotePage).all()} if page_id not in pages or set(order) != set(pages.keys()): return False pages[page_id].section = (section or "").strip() or None for i, pid in enumerate(order): pages[pid].position = i db.commit() return True def rename_section(db: DBSession, old_name: str, new_name: str | None) -> int: """あるセクションに属する全ページのセクション名を一括で変更する。new_name が空ならグループ解除になる。 戻り値: 変更したページ数。 """ normalized_new = (new_name or "").strip() or None pages = db.query(NotePage).filter(NotePage.section == old_name).order_by(NotePage.position, NotePage.id).all() for page in pages: page.section = normalized_new db.commit() return len(pages) def get_notes_context(db: DBSession) -> str | None: """共有ノートのタイトル一覧をシステムプロンプト用のテキストとして返す。セクションごとに見出しを付ける。""" pages = db.query(NotePage).order_by(NotePage.position, NotePage.id).all() lines = [ "【共有ノート】", "ユーザーとあなたが共同で編集できるノートです。以下のページがあります" "（セクションはページをグループ化する見出しで、あれば併記します）。", "", ] if pages: current_section = "__unset__" # どのページのsectionとも一致しない番兵 for page in pages: if page.section != current_section: current_section = page.section if current_section: lines.append(f"## {current_section}") preview = " ".join(page.content.split())[:40] suffix = f" — {preview}…" if preview else " — (空)" lines.append(f"- {page.title}{suffix}") else: lines.append("- (まだページがありません)") lines += [ "", "本文はここには含まれていません。内容が必要なときは read_note ツールで読み出してください。", "ノートへの記録・追記を頼まれたときは write_note ツールで書き込んでください" "（新しいタイトルを指定すればページが新規作成されます）。", "ユーザーが明示的に頼んでいないのに勝手に書き込まないこと。", ] return "\n".join(lines) def read_note(db: DBSession, title: str) -> str: """ページ本文をツール結果用のテキストとして返す。""" page = _find_page(db, title) if not page: available = ", ".join(p.title for p in db.query(NotePage).order_by(NotePage.position, NotePage.id).all()) return f"「{title}」というページはありません。現在のページ: {available or '(なし)'}" if not page.content.strip(): return f"ページ「{page.title}」は空です。" content = page.content if len(content) > NOTES_TOOL_CONTENT_LIMIT: content = content[:NOTES_TOOL_CONTENT_LIMIT] + "\n…(以下省略)" return f"ページ「{page.title}」の本文:\n\n{content}" _LIST_ITEM = re.compile(r"^\s*([-*+]|\d+[.)])\s") def _append(existing: str, addition: str) -> str: """既存本文に追記する。箇条書き同士をつなぐ場合だけ空行を挟まない。""" if not existing.strip(): return addition body = existing.rstrip() last_line = body.rsplit("\n", 1)[-1] first_line = addition.lstrip("\n").split("\n", 1)[0] separator = "\n" if _LIST_ITEM.match(last_line) and _LIST_ITEM.match(first_line) else "\n\n" return f"{body}{separator}{addition}" def write_note(db: DBSession, title: str, content: str, mode: str = "append") -> str: """ページに書き込む。存在しなければ新規作成する。""" if not title.strip(): return "タイトルが空のため書き込めませんでした。" page = _find_page(db, title) if page is None: page = NotePage(title=title.strip(), content=content, position=next_position(db)) db.add(page) db.commit() return f"新しいページ「{page.title}」を作成し、本文を書き込みました。" if mode == "replace": page.content = content result = f"ページ「{page.title}」の内容を置き換えました。" else: page.content = _append(page.content, content) result = f"ページ「{page.title}」に追記しました。" db.commit() return result def execute_note_tool(db: DBSession, name: str, args: dict) -> tuple[str, bool]: """ノート系ツールを実行する。戻り値: (ツール結果テキスト, ノートを変更したか)。""" try: if name == "read_note": return read_note(db, args.get("title", "")), False if name == "write_note": result = write_note( db, args.get("title", ""), args.get("content", ""), args.get("mode", "append"), ) return result, True except Exception as e: db.rollback() logger.exception("Note tool %s failed", name) return f"ノート操作に失敗しました: {e}", False return f"未知のツールです: {name}", False