Wikipedia 편집을 감시하는 로컬 AI 에이전트

def parse_sse_line(line: str) -> Optional[dict]:
    """SSE frames data as lines prefixed with 'data: '. Comment lines (starting with ':') and blank keep-alive lines are common on this feed and should be silently ignored, not treated as errors."""
    if not line or line.startswith(":"):
        return None
    if line.startswith("data:"):
        raw = line[len("data:"):].strip()
        if not raw:
            return None
        try:
            return json.loads(raw)
        except json.JSONDecodeError:
            return None
    return None

class Stage1Filter:
    """Wraps the velocity tracker and the byte-removal check into one pass/fail decision per event."""
    def __init__(self, tracker: Optional[EditVelocityTracker] = None):
        self.tracker = tracker or EditVelocityTracker()

    def evaluate(self, event: RecentChangeEvent) -> Optional[FilterSignal]:
        """Returns a FilterSignal if this event is worth the LLM's time, otherwise None, and None is the common case by a wide margin."""
        if event.is_bot:
            return None # bot edits have their own, separate review path
        recent_count = self.tracker.record_and_count(event.user, event.timestamp)
        bytes_removed = event.bytes_removed
        reasons = []
        if bytes_removed >= config.BYTES_REMOVED_THRESHOLD:
            reasons.append(f"removed {bytes_removed} bytes in one edit")
        if recent_count >= config.EDIT_VELOCITY_THRESHOLD:
            reasons.append(f"{recent_count} edits in {self.tracker.window_seconds}s")
        if not reasons:
            return None
        return FilterSignal(
            event=event,
            bytes_removed=bytes_removed,
            recent_edit_count=recent_count,
            reason="; ".join(reasons),
        )

async def evaluate_signal(signal: FilterSignal) -> AsyncIterator[str | AgentVerdict]:
    """Streams the model's raw output as it's generated (str chunks), then yields a final validated AgentVerdict once the stream completes. The caller tells the two apart with isinstance()."""
    client = ollama.AsyncClient(host=config.OLLAMA_HOST)
    stream = await client.chat(
        model=config.OLLAMA_MODEL,
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": _build_user_prompt(signal)},
        ],
        format=AgentVerdict.model_json_schema(),
        stream=True,
        options={"temperature": 0.1},
    )
    full_text = ""
    async for chunk in stream:
        piece = chunk["message"]["content"]
        full_text += piece
        if piece:
            yield piece # live token, for the broadcaster to forward immediately
    verdict = AgentVerdict.model_validate_json(full_text)
    yield verdict

async def publish(self, payload: dict) -> None:
    """Fans a payload out to every subscriber. A subscriber whose queue is full gets the message dropped rather than blocking the whole pipeline, a slow client should never be able to slow down the agent's actual processing loop."""
    message = json.dumps(payload)
    for queue in list(self._subscribers):
        try:
            queue.put_nowait(message)
        except asyncio.QueueFull:
            continue

async def run_pipeline() -> None:
    """Consumes the live stream forever, runs stage 1 on every event, and only calls the LLM stage on events that survive it."""
    async for event in wikipedia_event_stream():
        signal = stage1.evaluate(event)
        if signal is None:
            continue
        logger.info("Stage 1 flagged: %s by %s (%s)", signal.event.title, signal.event.user, signal.reason)
        await broadcaster.publish({"type": "flagged", "title": signal.event.title, "reason": signal.reason})
        try:
            async for item in evaluate_signal(signal):
                if isinstance(item, str):
                    await broadcaster.publish({"type": "token", "title": signal.event.title, "text": item})
                elif isinstance(item, AgentVerdict):
                    await broadcaster.publish({
                        "type": "verdict",
                        "title": signal.event.title,
                        "user": signal.event.user,
                        **item.model_dump(),
                    })
        except Exception:
            logger.exception("Stage 2 failed for %s, skipping this signal", signal.event.title)