Building a Streaming Local AI Agent

streaming-local-agent/
├── src/
│   ├── __init__.py
│   ├── config.py
│   ├── schemas.py
│   ├── stream_source.py
│   ├── filters.py
│   ├── agent.py
│   ├── broadcaster.py
│   └── main.py
├── tests/
│   └── test_filters.py
├── requirements.txt
└── .env.example

# src/stream_source.py
import asyncio
import json
import re
import time
from typing import AsyncIterator, Optional
import httpx

from .schemas import RecentChangeEvent
from . import config

# Wikipedia 在这个流中不会发送明确的 "用户是否匿名" 标志；
# 匿名编辑会使用编辑者的IP地址而非用户名进行归因，
# 因此在实际应用中，通过判断用户名是否为IP地址格式来识别匿名用户
_IPV4_RE = re.compile(r"^\d{1,3}(\.\d{1,3}){3}$")
_IPV6_RE = re.compile(r"^[0-9A-Fa-f:]+:[0-9A-Fa-f:]+$")

def is_anonymous_user(username: str) -> bool:
    return bool(_IPV4_RE.match(username) or _IPV6_RE.match(username))

def parse_sse_line(line: str) -> Optional[dict]:
    """解析SSE帧数据，这些数据以 'data: ' 开头。注释行
    （以 ':' 开头）和保持连接的空白行在该数据流中很常见，
    应该被静默忽略，而不是作为错误处理。"""
    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

def to_event(raw: dict) -> Optional[RecentChangeEvent]:
    """将原始的Wikimedia数据负载转换为我们的标准化模式。
    对于不关心的事件类型返回None而不是抛出异常，
    因为这个高流量的流会持续包含我们不关注的数据格式。"""
    if raw.get("type") != "edit":
        return None
    length = raw.get("length") or {}
    if "old" not in length or "new" not in length:
        return None
    return RecentChangeEvent(
        wiki=raw.get("wiki", "unknown"),
        user=raw.get("user", "unknown"),
        title=raw.get("title", "unknown"),
        is_anonymous=is_anonymous_user(raw.get("user", "")),
        is_bot=raw.get("bot", False),
        old_length=length["old"],
        new_length=length["new"],
        timestamp=raw.get("timestamp", time.time()),
        comment=raw.get("comment", "") or "",
    )

async def wikipedia_event_stream() -> AsyncIterator[RecentChangeEvent]:
    """被main.py使用的实时异步生成器。在连接断开时会自动重新连接，
    而不是因为一次网络故障就导致整个服务崩溃，这对于
    需要无人值守运行的服务非常重要。"""
    while True:
        try:
            async with httpx.AsyncClient(timeout=None) as client:
                async with client.stream("GET", config.WIKIPEDIA_STREAM_URL) as response:
                    async for line in response.aiter_lines():
                        raw = parse_sse_line(line)
                        if raw is None:
                            continue
                        if raw.get("wiki") not in config.WATCHED_WIKIS:
                            continue
                        event = to_event(raw)
                        if event is not None:
                            yield event
        except httpx.HTTPError:
            await asyncio.sleep(5)

# src/filters.py
import time
from collections import defaultdict, deque
from typing import Optional

from .schemas import RecentChangeEvent, FilterSignal
from . import config

class EditVelocityTracker:
    """按滑动窗口跟踪每个用户的最近编辑时间戳，使过滤器能够检测到快速连续的编辑爆发，而不仅仅是单次大范围删除。内存使用有限：旧用户会被移除，而非无限保存。"""

    def __init__(self, window_seconds: int = config.EDIT_VELOCITY_WINDOW_SECONDS,
                 max_tracked: int = config.MAX_TRACKED_WINDOWS):
        self.window_seconds = window_seconds
        self.max_tracked = max_tracked
        self._history: dict[str, deque[float]] = defaultdict(deque)

    def record_and_count(self, user: str, timestamp: float) -> int:
        """记录此次编辑并返回用户在滑动窗口内（包括本次）的编辑次数。"""
        history = self._history[user]
        history.append(timestamp)

        cutoff = timestamp - self.window_seconds
        while history and history[0] < cutoff:
            history.popleft()

        if len(self._history) > self.max_tracked:
            self._evict_oldest()

        return len(history)

    def _evict_oldest(self) -> None:
        oldest_user = min(self._history, key=lambda u: self._history[u][-1] if self._history[u] else 0)
        del self._history[oldest_user]

class Stage1Filter:
    """将速度跟踪器和字节删除检查整合为每个事件的通过/失败决策。"""

    def __init__(self, tracker: Optional[EditVelocityTracker] = None):
        self.tracker = tracker or EditVelocityTracker()

    def evaluate(self, event: RecentChangeEvent) -> Optional[FilterSignal]:
        """如果此事件值得LLM处理则返回FilterSignal，否则返回None。绝大多数事件都会返回None。"""
        if event.is_bot:
            return None  # 机器人编辑有独立的审核流程

        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"单次编辑删除了{bytes_removed}字节")
        if recent_count >= config.EDIT_VELOCITY_THRESHOLD:
            reasons.append(f"{recent_count}次编辑在{self.tracker.window_seconds}秒内")

        if not reasons:
            return None

        return FilterSignal(
            event=event, bytes_removed=bytes_removed,
            recent_edit_count=recent_count, reason="; ".join(reasons),
        )

# src/schemas.py
from __future__ import annotations
from pydantic import BaseModel, Field

class RecentChangeEvent(BaseModel):
    wiki: str
    user: str

    is_anonymous: bool
    is_bot: bool
    old_length: int
    new_length: int
    timestamp: float
    comment: str = ""

    @property
    def bytes_removed(self) -> int:
        return max(0, self.old_length - self.new_length)

class FilterSignal(BaseModel):
    event: RecentChangeEvent
    bytes_removed: int
    recent_edit_count: int
    reason: str

class AgentVerdict(BaseModel):
    """The structured judgment we force the local model to return.
    Constraining this with a schema is what makes the output usable in
    code rather than just readable by a human."""
    is_likely_vandalism: bool
    severity: int = Field(ge=1, le=5, description="1 = probably fine, 5 = high confidence vandalism")
    reasoning: str
    suggested_action: str

# src/agent.py
from typing import AsyncIterator
import ollama

from .schemas import FilterSignal, AgentVerdict
from . import config

SYSTEM_PROMPT = """You are a Wikipedia edit-monitoring assistant. You will be \
shown metadata about an edit that tripped an automated filter for a large \
deletion or unusually rapid editing. Decide whether this looks like likely \
vandalism or a legitimate edit (a rewrite, a cleanup, a merge). Respond with \
a JSON object matching the required schema. Be specific in your reasoning, \
reference the actual numbers you were given."""

def _build_user_prompt(signal: FilterSignal) -> str:
    e = signal.event
    return (
        f"Page: {e.title}\n"
        f"User: {e.user} ({'anonymous' if e.is_anonymous else 'registered'})\n"
        f"Bytes removed: {signal.bytes_removed}\n"
        f"Recent edit count by this user: {signal.recent_edit_count}\n"
        f"Edit summary left by user: \"{e.comment or '(none)'}\"\n"
        f"Trigger reason: {signal.reason}\n"
    )

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

# src/broadcaster.py
import asyncio
import json
from typing import AsyncIterator

class Broadcaster:
    def __init__(self, max_queue_size: int = 100):
        self._subscribers: set[asyncio.Queue] = set()
        self.max_queue_size = max_queue_size

    def subscribe(self) -> asyncio.Queue:
        queue: asyncio.Queue = asyncio.Queue(maxsize=self.max_queue_size)
        self._subscribers.add(queue)
        return queue

    def unsubscribe(self, queue: asyncio.Queue) -> None:
        self._subscribers.discard(queue)

    async def publish(self, payload: dict) -> None:
        """将负载分发给所有订阅者。当订阅者的队列已满时，消息会被丢弃而非阻塞整个管道，
        慢速客户端永远不应该能减缓代理的实际处理循环。"""
        message = json.dumps(payload)
        for queue in list(self._subscribers):
            try:
                queue.put_nowait(message)
            except asyncio.QueueFull:
                continue

    async def stream(self) -> AsyncIterator[str]:
        """调用者可以循环使用的异步生成器，由 main.py 中的 SSE 端点直接使用。"""
        queue = self.subscribe()
        try:
            while True:
                message = await queue.get()
                yield message
        finally:
            self.unsubscribe(queue)

# src/main.py
import asyncio
import logging
from contextlib import asynccontextmanager

from fastapi import FastAPI, Request
from sse_starlette.sse import EventSourceResponse

from .broadcaster import Broadcaster
from .filters import Stage1Filter
from .stream_source import wikipedia_event_stream
from .agent import evaluate_signal
from .schemas import AgentVerdict

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("streaming-local-agent")

broadcaster = Broadcaster()
stage1 = Stage1Filter()

async def run_pipeline() -> None:
    """无限消费实时流，在每个事件上运行阶段1过滤器，
    仅对通过阶段1的事件触发LLM处理阶段。"""
    async for event in wikipedia_event_stream():
        signal = stage1.evaluate(event)
        if signal is None:
            continue

        logger.info("阶段1标记: %s 由 %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("阶段2处理 %s 失败，跳过此信号", signal.event.title)

@asynccontextmanager
async def lifespan(app: FastAPI):
    task = asyncio.create_task(run_pipeline())
    logger.info("流式本地代理已启动，开始监控编辑...")
    yield
    task.cancel()
    logger.info("流式本地代理正在关闭")

app = FastAPI(title="Streaming Local Agent", lifespan=lifespan)

@app.get("/events")
async def events(request: Request):
    async def event_generator():
        async for message in broadcaster.stream():
            if await request.is_disconnected():
                break
            yield message
    return EventSourceResponse(event_generator())

ollama pull llama3.1:8b
ollama serve   # 如果尚未作为后台服务运行

python -m venv venv
source venv/bin/activate
pip install -r requirements.txt
uvicorn src.main:app --reload

curl -N http://localhost:8000/events