系列目录：[Phase 1-2 基础框架] → [Phase 3 状态机+文件读取] → [Phase 4-6 原生FC+Judge校验] → [Phase 5 多工具并行+BashTool] → [Phase 6 流式输出] → [Phase 7 Token预算与滑动窗口] → [Phase 8 摘要压缩] → [Phase 9 规划模式] → Phase 10 Web UI（本文）
上一篇链接：【Java/Go后端手撸原生Agent（第九篇）：Plan-and-Execute规划模式——从“走一步看一步“到“先谋后动“】

项目最终代码仓库

前言

前九篇做完，我们的Agent已经具备了：原生Function Calling、流式输出、多工具并行、Bash命令执行、LLM-as-Judge证据链校验、Token预算管理、滑动窗口+摘要压缩、Plan-and-Execute规划。它在命令行里能正确调用工具、能流式展示思考过程、长对话不会爆token、回答不全会被Judge打回、复杂任务会先做计划再执行。

但到目前为止，所有交互都在终端里。我们都是对着黑底白字看打字机效果，朋友、同事、产品经理看不到——CLI 是给开发者调试用的，不是给用户用的。一个真正可用的Agent，必须有一个用户界面。

本文做的事情很纯粹：把Agent核心从 print 耦合中彻底解放出来，变成一个纯事件生成器 run_agent_stream()；再用 FastAPI 暴露一个 SSE 端点，前端用单页深色 UI 通过 EventSource 消费事件流。最终你在浏览器里打开 http://localhost:8000 就能和Agent对话——计划卡片、工具卡片（可折叠）、打字机效果、Judge 徽章一应俱全，和命令行共用同一套Agent核心，零重复逻辑。

改造涉及三个文件：main.py（主循环→事件生成器）、app.py（FastAPI+SSE端点，新建）、web/index.html（单页UI，新建）。

一、问题本质：print 耦合让Agent长不出UI

前九篇的 run_agent() 函数长这样（简化）：

def run_agent(user_query, verbose=True):
    while state != FINISHED:
        if state == THINKING:
            for event in chat_completion_stream(messages, tools=tools_schema):
                if isinstance(event, ContentDelta):
                    print(event.delta, end="", flush=True)   # ← 直接print
                elif isinstance(event, ToolCallStart):
                    print(f"\n【工具调用】{event.name}")        # ← 直接print
                ...
            resp = ...
        elif state == TOOL_EXECUTING:
            obs = tool.execute(...)
            print(f"【工具返回】{obs[:200]}")                 # ← 直接print
        elif state == VALIDATING:
            print("【校验】正在Judge回答完整性...")            # ← 直接print

问题在哪？Agent的核心逻辑（状态机流转、LLM调用、工具执行、Judge校验）和"怎么把过程展示给人看"（print到stdout）耦合在同一个函数里。你想加一个Web UI？要么再写一份 run_agent_web() 复制90%的逻辑，要么硬着头皮在原函数里加 if/else 判断"当前是CLI模式还是Web模式"——两种做法都违反开闭原则。

第一性原理：Agent核心的产出不是"打印到终端的字符串"，而是**"发生了什么事"这个结构化事实**——“LLM输出了一段文字”、“LLM决定调用calculator”、“工具返回了结果”、“Judge通过了”、“任务完成”。至于这些事实怎么展示给人看（终端打字机、Web卡片、飞书消息、Slack Bot），是消费方的事，不是Agent核心的事。

这就是事件生成器模式的动机。

二、核心抽象：AgentEvent 事件模型

2.1 事件基类与自动序列化

先定义一个事件基类，所有具体事件继承它：

class AgentEvent:
    @staticmethod
    def _camel_to_snake(name: str) -> str:
        s1 = re.sub(r'(.)([A-Z][a-z]+)', r'\1_\2', name)
        return re.sub(r'([a-z0-9])([A-Z])', r'\1_\2', s1).lower()

    def to_sse(self) -> str:
        type_name = self._camel_to_snake(self.__class__.__name__.removesuffix("Event"))
        return json.dumps({"type": type_name, **self.payload()}, ensure_ascii=False)

    def payload(self) -> dict:
        return {}

两个设计要点：

1. 子类即类型：不需要手动维护事件类型枚举。ThinkingDeltaEvent 自动映射到 SSE 的 type: "thinking_delta"，加新事件只要继承 AgentEvent 并实现 payload()，前端自动能收到。
2. CamelCase→snake_case 自动转换：Python类名用大驼峰（ThinkingDeltaEvent），前端事件type用蛇形（thinking_delta），正则一次性转换。

⚠️ 踩坑记录：一开始图省事用 .lower() 转换，结果 ToolCallEvent 变成了 toolcall（下划线丢了），前端 case 'tool_call' 永远匹配不上。必须用正则处理驼峰边界：ToolCall → Tool_Call → tool_call。

2.2 9种具体事件

class PlanEvent(AgentEvent):
    def __init__(self, plan: dict): self.plan = plan
    def payload(self): return {"plan": self.plan}

class ThinkingDeltaEvent(AgentEvent):
    def __init__(self, delta: str): self.delta = delta
    def payload(self): return {"delta": self.delta}

class ToolCallEvent(AgentEvent):
    def __init__(self, name, args, tool_call_id): ...
    def payload(self): return {"name": self.name, "args": self.args, "id": self.tool_call_id}

class ToolResultEvent(AgentEvent):
    def __init__(self, name, result, is_error=False): ...
    def payload(self): return {"name": self.name, "result": self.result, "is_error": self.is_error}

class JudgeEvent(AgentEvent):
    def __init__(self, passed, missing="", retry_count=0): ...
    def payload(self): return {"passed": self.passed, "missing": self.missing, "retry_count": self.retry_count}

class ContextTrimEvent(AgentEvent):
    def __init__(self, blocks_removed, remaining_tokens, summary=""): ...
    def payload(self): return {"blocks_removed": ..., "remaining_tokens": ..., "summary": ...}

class DoneEvent(AgentEvent):
    def __init__(self, answer: str): self.answer = answer
    def payload(self): return {"answer": self.answer}

class ErrorEvent(AgentEvent):
    def __init__(self, message: str): ...
    def payload(self): return {"message": self.message}

class LoopEvent(AgentEvent):
    def __init__(self, round_num, state): ...
    def payload(self): return {"round": self.round_num, "state": self.state}

覆盖了Agent生命周期的所有关键节点：轮次开始（LoopEvent，用于前端状态栏显示"第N轮思考中"）、任务规划（PlanEvent）、LLM文本输出（ThinkingDeltaEvent，打字机增量）、工具调用/返回、Judge校验结果、上下文裁剪通知、错误、最终完成。

2.3 主循环改造：所有 print 替换为 yield

run_agent() 拆成两层：

• 底层 run_agent_stream(user_query) -> Generator[AgentEvent, None, str]：纯逻辑，零print，每个关键节点 yield XxxEvent(...)，最终 return final_answer。
• 上层 run_agent(user_query, verbose=True) -> str：CLI 适配层，消费生成器，按需要打印到终端，返回最终答案。

def run_agent_stream(user_query):
    # ... 初始化 memory/state/plan/tool_trace ...
    while state != FINISHED and loop_count < max_loop:
        if state == PLANNING:
            yield LoopEvent(round_num=loop_count, state="planning")
            task_plan = plan_task(user_query)
            yield PlanEvent(plan_dict)
            state = THINKING
        elif state == THINKING:
            yield LoopEvent(round_num=loop_count, state="thinking")
            for event in chat_completion_stream(messages, tools=tools_schema):
                if isinstance(event, ContentDelta):
                    yield ThinkingDeltaEvent(event.delta)        # ← yield事件
                elif isinstance(event, StreamDone):
                    resp = event.response
            if resp.has_tool_calls:
                state = TOOL_EXECUTING
            else:
                pending_draft_answer = resp.content
                state = VALIDATING
        elif state == TOOL_EXECUTING:
            yield LoopEvent(round_num=loop_count, state="tool_executing")
            for tc in pending_tool_calls:
                yield ToolCallEvent(name=tc.name, args=tc.arguments, tool_call_id=tc.id)
                obs = tool_map[tc.name].execute(tc.arguments)
                yield ToolResultEvent(name=tc.name, result=obs, is_error=is_error)
            state = THINKING
        elif state == VALIDATING:
            yield LoopEvent(round_num=loop_count, state="validating")
            judge_res = judge_answer(user_query, pending_draft_answer, tool_trace, task_plan)
            if judge_res.passed or judge_retry_count >= max_judge_retries:
                yield JudgeEvent(passed=True, missing="", retry_count=judge_retry_count)
                final_answer = pending_draft_answer
                state = FINISHED
            else:
                yield JudgeEvent(passed=False, missing=judge_res.missing, retry_count=judge_retry_count)
                state = THINKING
    yield DoneEvent(final_answer)
    return final_answer

CLI 包装器变得非常干净：

def run_agent(user_query, verbose=True):
    answer = ""
    for event in run_agent_stream(user_query):
        if verbose:
            if isinstance(event, ThinkingDeltaEvent):
                print(event.delta, end="", flush=True)
            elif isinstance(event, ToolCallEvent):
                print(f"\n【工具调用】{event.name}({json.dumps(event.args, ensure_ascii=False)})")
            elif isinstance(event, JudgeEvent):
                print("【校验通过】回答完整" if event.passed else f"【门禁拦截】缺失: {event.missing}")
            ...
        if isinstance(event, DoneEvent):
            answer = event.answer
    return answer

关键收益：Agent核心不关心自己被谁消费。CLI是一个消费者，Web UI是另一个消费者，未来要加飞书/Slack Bot、Discord Bot、微信小程序，都是再写一个消费方，Agent核心零改动。

三、FastAPI + SSE 端点

新建 app.py，非常薄——核心逻辑全部在 main.py，这层只做协议转换：

from fastapi import FastAPI, Query
from fastapi.responses import StreamingResponse, FileResponse
from main import run_agent_stream, AgentEvent

app = FastAPI(title="Native Agent Demo")
WEB_DIR = Path(__file__).parent / "web"

def sse_generator(query: str):
    try:
        for event in run_agent_stream(query):
            data = event.to_sse()
            yield f"data: {data}\n\n"
    except Exception as e:
        err = json.dumps({"type": "error", "message": f"{type(e).__name__}: {e}"}, ensure_ascii=False)
        yield f"data: {err}\n\n"
    finally:
        yield "data: [DONE]\n\n"

@app.get("/api/chat")
def chat(query: str = Query(...)):
    return StreamingResponse(
        sse_generator(query),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",    # ← 关键：禁用Nginx/反向代理缓冲
        },
    )

@app.get("/")
def index():
    return FileResponse(WEB_DIR / "index.html")

SSE 协议极简：

• 响应头 Content-Type: text/event-stream
• 每条消息格式 data: {json}\n\n（两个换行分隔）
• 结束信号 data: [DONE]\n\n
• X-Accel-Buffering: no 是必要的——否则反向代理会缓冲整个响应，SSE 退化成"一次性返回"，打字机效果消失。

⚠️ 踩坑记录：一开始服务起了，但前端收到事件是"等好几秒一起砸过来"，没有打字机效果。排查了半天发现不是代码问题，是浏览器或代理的缓冲。最终在响应头里加了 X-Accel-Buffering: no，本地 uvicorn 直连没有缓冲问题，但养成习惯加上，防止部署到Nginx后面失效。

启动命令：python3 -m uvicorn app:app --host 0.0.0.0 --port 8000。

四、前端单页 UI：深色主题 + EventSource 消费

前端是一个纯 HTML+CSS+JS 单页（无框架、无构建工具），核心是通过浏览器原生 EventSource API 消费SSE流：

const url = '/api/chat?query=' + encodeURIComponent(query);
eventSource = new EventSource(url);

eventSource.onmessage = (e) => {
    if (e.data === '[DONE]') {
        eventSource.close();
        finishTurn();
        return;
    }
    const evt = JSON.parse(e.data);
    handleEvent(evt);
};

handleEvent 做事件分发：

function handleEvent(evt) {
    switch (evt.type) {
        case 'plan':          renderPlan(evt.plan); break;
        case 'thinking_delta':appendContent(evt.delta); break;
        case 'tool_call':     createToolCard(evt); break;
        case 'tool_result':   updateToolResult(evt); break;
        case 'judge':         renderJudge(evt); break;
        case 'error':         addSysMsg('❌ ' + evt.message, 'error'); break;
        case 'done':          renderDone(evt.answer); break;
        case 'loop':          updateStatus(evt); break;
        ...
    }
}

4.1 视觉组件

• 用户气泡：右侧蓝色气泡，iMessage风格
• Assistant气泡：左侧深色卡片，内含多个组件
• Plan卡片：紫色左边框，显示目标+步骤列表+目的
• Tool卡片：可折叠（点击header展开/收起），默认折叠，header显示工具名和状态（执行中/完成/错误），展开后显示参数和返回结果
• Judge徽章：绿色"✓ Judge 校验通过"或红色"✗ Judge 打回: 缺少xxx"
• 系统消息：上下文裁剪、重复调用拦截、错误提示等

4.2 打字机光标

用CSS动画实现闪烁光标：

.typing-cursor {
    display: inline-block;
    width: 2px; height: 1em;
    background: var(--accent);
    animation: blink 0.8s infinite;
}
@keyframes blink { 0%,100%{opacity:1} 50%{opacity:0} }

每次 thinking_delta 追加文本后，调用 moveCursorToEnd() 把光标元素移到气泡末尾，保证光标始终在打字位置。

五、演进修正：从"双份文本"到"草稿归档"

UI 跑起来第一版就发现一个问题（见题图的修复过程）：回答结束后同一段文本显示了两次——流式输出打了一遍，done 事件里 answer 字段又贴了一遍。

5.1 根因

不是 bug，是事件契约的语义重叠：

• THINKING阶段的 ThinkingDeltaEvent 把 resp.content 逐字增量发给前端 → 前端累加到 .answer-text
• VALIDATING 通过后 yield DoneEvent(final_answer)，而 final_answer = pending_draft_answer = resp.content —— 同一个字符串
• 前端在 renderDone 里新建 .final-answer 容器又贴了一遍

一开始想到的修法是"judge通过后把第一段折叠起来"——但这只是遮羞：同一份信息用两个容器承载，折叠只是把重复隐藏了。第一性原理追问：为什么需要两个容器？

答案是不需要。流式输出本身就是最终答案的生长过程，DoneEvent 的 answer 字段是给CLI return 用的（同步返回值），前端不需要再次渲染——流式已经打完了，done 只是"收尾信号"。

5.2 更深层的问题：judge打回时多轮草稿糊在一起

修完重复问题后又发现第二个隐藏bug：judge打回时（回答不完整），Agent会带着Judge反馈进入下一轮THINKING，新一轮的 thinking_delta 会追加到同一个DOM容器——草稿和终稿糊在一起，无法区分哪段是被打回的草稿、哪段是最终回答。

5.3 最终方案：草稿归档+分段隔离

改前端渲染逻辑，不改动后端契约：

1. 每轮THINKING独立段：用 currentDraft 指针跟踪当前正在写入的 .answer-text 容器；新一轮开始（工具返回后、judge打回后）自动创建新段，段间加分隔线。
2. Judge打回的草稿归档：judge fail时给草稿段加 .rejected 类（45%透明度、60px高度、渐变遮罩折叠、右下角"草稿，已被Judge打回，点击展开"），点击可展开/收起；新段在下方重新打字。
3. renderDone 不再渲染answer文本：流式打字出来的那段就是最终答案，done 退化为纯收尾信号。
4. 工具切换点自动封段：tool_call 到来时把"我来调用xxx工具"这类过渡旁白封存，工具卡片在段后展示；工具返回后新思考自动开新段。

.answer-text.rejected {
    opacity: 0.45;
    max-height: 60px;
    overflow: hidden;
    mask-image: linear-gradient(to bottom, #000 30%, transparent 100%);
    cursor: pointer;
}
.answer-text.rejected::after {
    content: '（草稿，已被 Judge 打回，点击展开）';
    position: absolute; bottom: 0; right: 0;
    background: var(--surface);
    color: var(--text-dim);
    font-size: 11px; padding: 2px 6px; border-radius: 4px;
}

let currentDraft = null;  // 当前正在写入的草稿段指针

function appendContent(delta) {
    if (!currentDraft) {
        const hasPrior = currentContentDiv.querySelector('.answer-text') !== null;
        if (hasPrior) {
            const sep = document.createElement('div');
            sep.className = 'draft-sep';
            currentContentDiv.appendChild(sep);
        }
        currentDraft = document.createElement('div');
        currentDraft.className = 'answer-text';
        currentContentDiv.appendChild(currentDraft);
        moveCursorToEnd();
    }
    currentDraft.textContent += delta;
}

function rejectDraft() {
    if (!currentDraft) return;
    currentDraft.classList.add('rejected');
    currentDraft.addEventListener('click', () => currentDraft.classList.toggle('expanded'));
    currentDraft = null;
}

function renderDone(answer) {
    if (currentDraft) currentDraft.classList.add('final');
    // 不再创建新的 .final-answer 容器
}

这个方案实现了三个目标：

• 无重复文本（单容器承载最终答案）
• 草稿不丢（可点击展开查看Judge打回的历史草稿，作为调试证据）
• 终稿干净（草稿默认折叠灰显，不干扰终稿阅读）

这是个重要的工程经验：UI问题先查事件契约的语义是否清晰。"两段一样的文字"第一眼像前端渲染bug，根因是后端事件模型里同一份数据通过两个通道发送。先追契约，再改渲染，不要用折叠/隐藏掩盖语义重叠。

六、.env 加载路径 Bug

开发中还踩了一个小坑，值得记录：load_dotenv() 默认依赖当前工作目录（CWD）。在项目根目录执行 python3 -m uvicorn app:app 没问题，但如果从别的目录启动（比如 cd /tmp && python3 /path/to/uvicorn app:app），.env 文件找不到，API_KEY为空，所有LLM调用401。

修复原则：环境加载必须基于 __file__ 的绝对路径，不能依赖CWD。

# env_loader.py
from pathlib import Path
from dotenv import load_dotenv

_ENV_PATH = Path(__file__).parent.parent / ".env"   # ← 基于本文件位置
load_dotenv(_ENV_PATH)

class LLMConfig:
    API_KEY = os.getenv("OPENAI_API_KEY", "")
    BASE_URL = os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
    MODEL_NAME = os.getenv("MODEL_NAME", "gpt-4o-mini")

七、整体架构回顾

至此整个Agent的分层结构变得非常清晰：

┌─────────────────────────────────────────────┐
│          消费方（展示层，可插拔）              │
│  ┌──────────┐  ┌──────────┐  ┌───────────┐  │
│  │ CLI(run_ │  │ Web UI   │  │ 未来:飞书/ │  │
│  │ agent)   │  │ (SSE+JS) │  │ Slack Bot │  │
│  └────┬─────┘  └────┬─────┘  └─────┬─────┘  │
│       │             │              │        │
│       └─────────────┼──────────────┘        │
│                     │ 消费事件流              │
├─────────────────────┼───────────────────────┤
│                     ▼                        │
│  ┌──────────────────────────────────────┐   │
│  │  run_agent_stream()  事件生成器        │   │
│  │  (状态机+LLM+工具+Judge+Token管理)    │   │
│  │  纯逻辑，零IO依赖                     │   │
│  └──────────────────────────────────────┘   │
│                     │ yield AgentEvent       │
│                     ▼                        │
│  ┌──────────────────────────────────────┐   │
│  │  基础设施层                           │   │
│  │  llm_client / tools / memory / ...   │   │
│  └──────────────────────────────────────┘   │
└─────────────────────────────────────────────┘

Agent核心是一个纯函数（生成器），输入是用户query，输出是事件流。展示层是可插拔的适配器，新的展示渠道不需要改核心逻辑。

八、最终效果

启动服务：

cd native-agent-demo
python3 -m uvicorn app:app --host 0.0.0.0 --port 8000

浏览器打开 http://localhost:8000，你能看到：

• 深色主题单页UI，顶部状态栏实时显示Agent状态（规划中/第N轮思考中/执行工具/校验中/就绪）
• Plan卡片展示任务目标和步骤列表（紫色左边框）
• 工具卡片默认折叠（点击header展开看参数和返回），header实时显示"执行中/完成/错误"状态
• 正文打字机效果+闪烁光标
• Judge通过时显示绿色徽章，打回时红色徽章+草稿灰显折叠
• 上下文裁剪、重复调用拦截有系统提示
• 支持Enter发送/Shift+Enter换行，输入框自动高度

可以试几个query：

• 点击建议chip “计算 (38+42)*5/3”（简单单工具调用）
• “列出当前目录下所有文件”（bash工具）
• “读取 main.py 的前50行并解释”（read_file工具）
• 多步骤组合任务，触发Plan-and-Execute全流程

九、小结

本文做的事情：

1. 解耦：把Agent主循环从print耦合重构为纯事件生成器 run_agent_stream()，9种AgentEvent覆盖生命周期所有关键节点。
2. 协议层：FastAPI + SSE端点，把事件流转为标准SSE协议，支持浏览器端实时消费。
3. 展示层：单页深色Web UI，通过EventSource消费事件流，实现Plan卡片、工具卡片（可折叠）、打字机效果、Judge徽章、草稿归档。
4. 契约修正：修复了事件语义重叠导致的"双份文本"问题和多轮草稿糊在一起的问题，实现judge打回草稿的灰显折叠归档。
5. 健壮性：修复了 .env 路径依赖CWD的bug，事件type CamelCase→snake_case转换用正则而非lower()。

改造后的Agent有了真正的用户界面，CLI降级为开发者调试工具。更重要的是，分层架构清晰——核心逻辑与展示层解耦，未来扩展新的展示渠道（飞书/Slack/Discord/小程序）只需要写新的适配器，核心零改动。

下一篇我们考虑做什么？可以做长期记忆（向量数据库RAG），可以做多Agent协作，也可以做工具的自动发现与注册。