weft

Python Agent Runtime — long-lived session orchestration, event replay, ask-user interrupt/resume.

weft 把"一次 HTTP 调用"模型升级为长生命周期、可订阅、可重连的 Agent 会话。Transport 层和 Agent 框架解耦——LangGraph / OpenAI Agents SDK / 裸 LLM SDK 都能接入。

核心能力

跟生态的关系

30 秒 demo

import asyncio
from weft import (
    AskUserHandler, CancelToken, EventEmitter,
    MainBlockStart, MainBlockDelta, MainBlockEnd,
    ThreadRunner, WSListener,
)


class EchoAgent:
    async def run_turn(
        self, user_input: str, *,
        emit: EventEmitter, askuser: AskUserHandler, cancel_token: CancelToken,
    ) -> None:
        await emit(MainBlockStart(block_id="b1", block_type="text"))
        for ch in user_input:
            if cancel_token.cancelled:
                break
            await emit(MainBlockDelta(block_id="b1", delta=ch))
        await emit(MainBlockEnd(block_id="b1"))


async def main():
    runner = ThreadRunner("t-1", EchoAgent())

    captured = []
    async def send(payload): captured.append(payload)

    listener = WSListener(send)
    await runner.attach_listener(listener)
    await runner.start_turn("hi", listener)
    assert runner._turn_task is not None
    await runner._turn_task

asyncio.run(main())

完整 WS server demo:examples/hello_echo/。

FastAPI 适配器

weft.adapters.fastapi.run_ws_session 把上面那段"收 client → 路由到 runner"接收循环抽成一行调用; 业务侧扩展走鸭子类型 WSSessionHooks,全部方法可选,不实现等于 no-op。

from fastapi import FastAPI, WebSocket
from weft import RunnerJanitor, RunnerRegistry
from weft.adapters.fastapi import run_ws_session

app = FastAPI()
registry = RunnerRegistry(my_agent_factory)
janitor = RunnerJanitor(registry); janitor.start()

class Hooks:
    async def on_attach(self, listener):
        listener.enqueue(my_usage_snapshot())   # 补一帧客户端 hydrate 用的快照
    async def on_resume_submitted(self, answers):
        await persist_clarify(answers)          # clarify 答案落业务库
    async def on_config(self, msg):
        await update_role_models(msg)           # ConfigUpdate 业务字段由 hook 解释

@app.websocket("/ws/{thread_id}")
async def ws(ws: WebSocket, thread_id: str):
    runner = await registry.get_or_create(thread_id)
    await run_ws_session(ws, thread_id, runner, hooks=Hooks())

run_ws_session 负责: ws.accept / 推 ReadyEvent / 收 hello 触发 attach 补帧 / 路由 user_message / cancel / resume / compact 到 runner / detach 关 listener / pump 死亡时主动 close ws。安装: pip install "weft[fastapi]"。

架构

┌─────────────────────────────────────────────────────────────┐
│  Transport adapter (FastAPI WS / SSE / 自定义)             │
│  ↓ attach_listener  ↑ user_message/resume/cancel            │
├─────────────────────────────────────────────────────────────┤
│  ThreadRunner                                              │
│  ├─ state machine: idle / streaming / awaiting_resume       │
│  ├─ ring buffer + compute_replay (last_seq 补帧)            │
│  ├─ WSListener[] (per-transport queue + pump, 背压隔离)     │
│  └─ ask-user future / cancel salvage                        │
├─────────────────────────────────────────────────────────────┤
│  AgentProtocol  (你的实现, 或 weft.adapters.langgraph)      │
│  └─ run_turn(emit, askuser, cancel_token)                   │
└─────────────────────────────────────────────────────────────┘

RunnerRegistry 按 key 索引 runner,RunnerJanitor 周期回收 idle。

安装

pip install weft                    # 核心 (零 langgraph/langchain 依赖)
pip install "weft[langchain]"       # + LangGraph adapter (规划中)
pip install "weft[fastapi]"         # + FastAPI WS adapter (规划中)

Status

v0.0.1 — alpha。核心 transport/lifecycle 层稳定,middleware 套装 + LangGraph adapter 在 v0.1 完成。

详细设计见 docs/architecture.md,事件协议见 docs/protocol.md。

License

MIT