"""庫柏 SDK 路徑 — LINE 訊息進來的入口。 階段 10-B 重構: - 不再每次 LINE 訊息 spawn claude CLI 子程序(75s) - 改用 cooper_client_manager 的全域 ClaudeSDKClient(暖機 + 常駐,~5-10s) - 工具呼叫的 requester_id 透過 ContextVar 傳(取代每次重建 closure) - session_id = "line:" 區分多群組對話狀態 行為對 caller 透明:跟舊版一樣 signature。caller 不用改。 """ from __future__ import annotations import logging from datetime import date from uuid import UUID from sqlalchemy import select from app.config import settings from app.core.ws_manager import manager as ws_manager from app.database import AsyncSessionLocal from app.models.channel import Message from app.services.coba_chat import ( detect_name_match, detect_coba_mention, is_in_engagement_db, mark_engagement, _build_messages_context, _build_rag_context, _build_system_prompt, _save_coba_message, _load_line_image_b64, ) logger = logging.getLogger("coba.ai.sdk") def _flatten_history_to_prompt(claude_messages: list[dict]) -> str: """把 anthropic-style messages list 壓成 SDK 用的 single string prompt。 每則訊息一行,有 [姓名] prefix。圖片暫時 fallback 成「(對方傳了一張圖)」。 """ lines = [] for m in claude_messages: role = m.get("role", "user") content = m.get("content", "") if isinstance(content, list): text_parts = [] for block in content: if isinstance(block, dict): if block.get("type") == "text": text_parts.append(block.get("text", "")) elif block.get("type") == "image": text_parts.append("(對方傳了一張圖,你目前看不到實際內容)") content = "\n".join(text_parts) if role == "assistant": lines.append(f"[庫柏 自己之前說] {content}") else: lines.append(content) return "\n\n".join(lines) async def coba_respond_to_mention_sdk(channel_id: UUID, triggered_by_message_id: UUID) -> None: """SDK 版本入口:用全域常駐 ClaudeSDKClient 跑 Cooper 對話。""" if not settings.CLAUDE_CODE_OAUTH_TOKEN: logger.warning("CLAUDE_CODE_OAUTH_TOKEN 沒設,跳過庫柏 SDK 回應") return # 動態 import — 避免啟動時 pull SDK 連 module-level 動作 from app.services.cooper_client import cooper_client_manager # 廣播「庫柏正在輸入」 await ws_manager.broadcast({ "type": "coba_typing", "channel_id": str(channel_id), }) response_text = "(庫柏沒說話)" rate_limit_type: str | None = None rate_limit_status: str | None = None total_cost_usd = 0.0 tool_calls: list[dict] = [] requester_id: UUID | None = None async with AsyncSessionLocal() as db: try: # 1. 抓 triggered message + requester triggered_msg = (await db.execute( select(Message).where(Message.id == triggered_by_message_id) )).scalar_one_or_none() requester_id = triggered_msg.user_id if triggered_msg and triggered_msg.user_id else None if requester_id is None: response_text = "(庫柏無法判定誰下令,先不執行)" msg = await _save_coba_message(db, channel_id, response_text) await _broadcast_msg(db, channel_id, msg) return # 2. 組 dynamic prompt context # system prompt 已經在 cooper_client startup 時用「靜態角色 COBA_BASE_PROMPT」設過了。 # 這裡只把「動態」部分塞進 user prompt 開頭(日期、團隊清單、近 10 則訊息、RAG)。 from app.models.user import User users = list((await db.execute( select(User).where(User.is_active.is_(True), User.deleted_at.is_(None)) )).scalars().all()) team_lines = "\n".join(f"- {u.display_name}({u.role})" for u in users) history_messages = await _build_messages_context(db, channel_id) history_prompt = _flatten_history_to_prompt(history_messages) rag_text = "" if triggered_msg: rag_text = await _build_rag_context(channel_id, triggered_msg.content) dynamic_prefix = ( f"【今日日期】{date.today().isoformat()}\n\n" f"【目前團隊成員】\n{team_lines}\n" f"{rag_text}" ) # 階段 10-H:抓近期 LINE 圖片,build 成 multimodal content blocks。 # 規則:近 10 則訊息內,message_type='image' 的最多挑 3 張(從新到舊), # 每張 base64 < 4.5MB(Anthropic 5MB 限制留 buffer)。 image_blocks: list[dict] = [] try: recent_image_msgs = (await db.execute( select(Message).where( Message.channel_id == channel_id, Message.deleted_at.is_(None), Message.message_type == "image", ).order_by(Message.created_at.desc()).limit(10) )).scalars().all() for m in recent_image_msgs: if len(image_blocks) >= 3: break block = _load_line_image_b64(m.line_message_id) if block: image_blocks.append(block) if image_blocks: image_blocks.reverse() # 由舊到新,跟對話順序一致 image_note = ( f"\n\n【附圖】我會附上群組裡近期 {len(image_blocks)} 張圖讓你直接看," f"請對圖片內容做具體回應(不要說「我看不到」)。" ) full_prompt = f"{dynamic_prefix}\n\n【近期對話】\n{history_prompt}{image_note}" else: full_prompt = f"{dynamic_prefix}\n\n【近期對話】\n{history_prompt}" except Exception: logger.exception("build image blocks 失敗,降級成純文字 prompt") image_blocks = [] full_prompt = f"{dynamic_prefix}\n\n【近期對話】\n{history_prompt}" # 3. 呼叫全域 client(每群組獨立 session_id) session_id = f"line:{channel_id}" result = await cooper_client_manager.respond( prompt_text=full_prompt, session_id=session_id, requester_id=requester_id, image_blocks=image_blocks or None, ) response_text = result["response_text"] rate_limit_type = result["rate_limit_type"] rate_limit_status = result["rate_limit_status"] total_cost_usd = result["total_cost_usd"] tool_calls = result["tool_calls"] # 4. 寫 AgentActivity(billing 證據 — 看 rate_limit_type) try: from app.models.agent_activity import AgentActivity async with AsyncSessionLocal() as _adb: _adb.add(AgentActivity( activity_type="line_response", user_id=requester_id, summary=response_text[:120] if response_text else "", success=True, extra={ "channel_id": str(channel_id), "model": "claude-sonnet-4-5", "tool_calls": tool_calls, "images_sent": len(image_blocks), "via": "claude_agent_sdk_persistent", "rate_limit_type": rate_limit_type, "rate_limit_status": rate_limit_status, "billing": ( "max_subscription_5h_pool" if rate_limit_type == "five_hour" else ( # 同一 client connect 只在第一次 query 發 RateLimitEvent。 # 之後 None 不代表降級 — 繼承同一個 connection 的 billing 來源。 "max_subscription_5h_pool_inherited" if rate_limit_type is None else f"unknown:{rate_limit_type}" ) ), "sdk_inference_cost_usd": total_cost_usd, }, cost_usd=0.0, )) await _adb.commit() except Exception: logger.exception("寫 AgentActivity (line_response sdk) 失敗") except Exception as e: import traceback tb = traceback.format_exc() print(f"[COOPER_SDK_FAIL] {type(e).__name__}: {e}\n{tb}", flush=True) logger.exception("Cooper SDK 失敗") response_text = f"(庫柏現在抽風了,稍後再試:{type(e).__name__})" # 5. 寫 DB + 廣播 + 標記 engagement msg = await _save_coba_message(db, channel_id, response_text) if triggered_msg and triggered_msg.user_id: mark_engagement(channel_id, triggered_msg.user_id) await _broadcast_msg(db, channel_id, msg) async def _broadcast_msg(db, channel_id: UUID, msg) -> None: from app.api.messages import ( _build_read_by, _get_channel_read_states, _serialize_message, _ws_safe, ) member_reads = await _get_channel_read_states(channel_id, db) serialized = _serialize_message(msg, _build_read_by(msg, member_reads)) await ws_manager.broadcast(_ws_safe({ "type": "message_new", "channel_id": str(channel_id), "message": serialized, })) # 跟舊版一樣的 signature(line_bridge / messages.py 不用改) import asyncio def trigger_coba_if_mentioned_sdk(channel_id: UUID, message_id: UUID, content: str, user_id: UUID | None = None) -> None: if detect_name_match(content): asyncio.create_task(coba_respond_to_mention_sdk(channel_id, message_id)) return if user_id is None: return asyncio.create_task(_check_engagement_then_respond_sdk(channel_id, message_id, user_id)) async def _check_engagement_then_respond_sdk(channel_id: UUID, message_id: UUID, user_id: UUID) -> None: async with AsyncSessionLocal() as db: if await is_in_engagement_db(channel_id, user_id, db): await coba_respond_to_mention_sdk(channel_id, message_id)