4eeddc4b1f
Modular async pipeline: FrameSource->Vision->context and STT/text->Brain->TTS. All stages are Protocols; mock backends run end-to-end with no deps/keys. Real backends included: mss screen capture, Claude vision+brain (guarded imports).
82 lines
3.0 KiB
Python
82 lines
3.0 KiB
Python
"""Cloud brain + vision via the Anthropic (Claude) API.
|
|
|
|
Both share one client. Vision sends the frame as a base64 image; the brain is a
|
|
plain chat call that receives the latest screen description as context.
|
|
|
|
Requires: pip install anthropic
|
|
Env: ANTHROPIC_API_KEY
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import os
|
|
import time
|
|
|
|
from ..interfaces import Frame, Reply, ScreenObservation
|
|
|
|
|
|
def _client(api_key: str | None):
|
|
import anthropic
|
|
|
|
return anthropic.AsyncAnthropic(api_key=api_key or os.environ.get("ANTHROPIC_API_KEY"))
|
|
|
|
|
|
class ClaudeVision:
|
|
def __init__(self, *, model: str = "claude-sonnet-4-5", api_key: str | None = None) -> None:
|
|
self.model = model
|
|
self._client = _client(api_key)
|
|
|
|
async def describe(self, frame: Frame, hint: str | None = None) -> ScreenObservation:
|
|
prompt = hint or (
|
|
"이건 디스코드 화면공유 캡처야. 지금 화면에서 무슨 일이 벌어지는지 "
|
|
"2~3문장으로 한국어로 간결하게 설명해줘. 코드/에러/게임/문서 등 맥락을 짚어줘."
|
|
)
|
|
b64 = base64.b64encode(frame.data).decode()
|
|
resp = await self._client.messages.create(
|
|
model=self.model,
|
|
max_tokens=300,
|
|
messages=[
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{
|
|
"type": "image",
|
|
"source": {"type": "base64", "media_type": frame.mime, "data": b64},
|
|
},
|
|
{"type": "text", "text": prompt},
|
|
],
|
|
}
|
|
],
|
|
)
|
|
text = "".join(b.text for b in resp.content if b.type == "text")
|
|
return ScreenObservation(text=text.strip(), ts=frame.ts)
|
|
|
|
|
|
class ClaudeBrain:
|
|
SYSTEM = (
|
|
"너는 사용자의 디스코드 화면공유를 실시간으로 함께 보는 AI 파트너야. "
|
|
"화면 설명을 참고해 자연스러운 반말/존댓말은 사용자에 맞추고, 짧고 대화하듯 답해. "
|
|
"화면을 못 봤으면 솔직히 말해."
|
|
)
|
|
|
|
def __init__(self, *, model: str = "claude-sonnet-4-5", api_key: str | None = None) -> None:
|
|
self.model = model
|
|
self._client = _client(api_key)
|
|
|
|
async def respond(self, user_text, screen, history) -> Reply:
|
|
msgs = []
|
|
for user, ai in history:
|
|
msgs.append({"role": "user", "content": user})
|
|
msgs.append({"role": "assistant", "content": ai})
|
|
screen_note = f"[지금 화면] {screen.text}\n\n" if screen else "[지금 화면] (아직 못 읽음)\n\n"
|
|
msgs.append({"role": "user", "content": screen_note + user_text})
|
|
resp = await self._client.messages.create(
|
|
model=self.model,
|
|
max_tokens=400,
|
|
system=self.SYSTEM,
|
|
messages=msgs,
|
|
)
|
|
text = "".join(b.text for b in resp.content if b.type == "text")
|
|
return Reply(text=text.strip(), ts=time.monotonic())
|