系列Java 工程师转 AI Agent 3 个月学习计划作者宸丶一| 28 岁 Java 程序员规划狂魔周日肝了一整天今日目标从 yield 生成器到完整 ChatGPT 风格的 Web 聊天应用个人格言代码改不改变世界我不知道但先让我准时下班。前言大家好我是宸一一个28岁的Java程序员。今天是第8天周日本来计划只学一个流式输出结果越写越上头直接从 yield 生成器一路撸到了一个完整的 Web 聊天应用。有侧边栏、有多会话管理、有流式打字效果、有工具调用、有 SQLite 持久化——本质上就是个迷你豆包。先看成果是的一个 Java 后端用 Python 手撸了一个 AI 聊天应用。虽然 UI 简陋了点但核心架构和 ChatGPT、豆包是一样的。一、今日学习路线01 流式基础yield 生成器02 SSE协议FastAPI 流式接口03 测试客户端Python 验证04 真实APIMiMo 流式调用05 工具调用流式Tool Calling06 持久化SQLite 会话管理07 Web应用手撸豆包从 yield 到完整 Web 应用7 步走完。二、为什么需要流式输出2.1 普通模式 vs 流式模式普通模式Day 7 用户发消息 → 等3秒 → 一大坨文字砸脸上 用户内心这破程序卡住了 流式模式Day 8 用户发消息 → 你 → 你好 → 你好 → 边打边看 用户内心它在思考了在回答了LLM 天生就是逐字生成的。它并不是先想好完整答案再吐出来而是每一步只预测下一个 token。所以流式输出 拿到一个 token 就立刻返回一个。2.2 Python 生成器yield流式输出的核心是 Python 的yield关键字。作为 Java 后端你可以把它理解为 Iterator 的语法糖defgenerate_numbers():foriinrange(1,6):print(f 正在生成第{i}个...)yieldi# 返回值 暂停下次继续# 调用后不执行返回生成器对象gengenerate_numbers()next(gen)# → 1next(gen)# → 2# for 循环自动调用 nextfornumingenerate_numbers():print(num)JavaPythonIteratorTGeneratorhasNext()自动 StopIterationnext()next()或for要写一个类实现接口一个yield搞定Python 的 yield 比 Java 简洁太多了。Java 要写一个类实现 Iterator 接口Python 只要一个 yield 关键字。三、SSE 协议 服务器推送3.1 什么是 SSESSEServer-Sent Events就是服务器持续往客户端推数据连接不断开。HTTP 普通请求一问一答答完断开 浏览器 → 请求 → 服务器 → 完整响应 → 断开 SSE服务器持续推送 浏览器 → 请求 → 服务器 → chunk1 → 浏览器显示 → chunk2 → 浏览器显示 → chunk3 → 浏览器显示 → [DONE] → 结束Java 对照SSE Spring 的SseEmitter。原理一模一样只是 Python 写法不同。3.2 FastAPI 实现 SSEfromfastapi.responsesimportStreamingResponseapp.post(/chat/stream)defchat_stream(request:ChatRequest):defevent_generator():fortokeninllm_stream(request.message):yieldfdata: {{content: {token}}}\n\nyielddata: [DONE]\n\nreturnStreamingResponse(event_generator(),media_typetext/event-stream# SSE 的 MIME 类型)一行StreamingResponse就把生成器变成 HTTP 流。Java 写 SseEmitter 要多少行3.3 SSE 数据格式data: {content: 你}\n\n data: {content: 好}\n\n data: [DONE]\n\n每个data:后面跟一个 JSON\n\n是事件分隔符。[DONE]表示流结束。四、真实 MiMo API 流式调用前面用的都是假数据mock现在接入真实的大模型 APIdefmimo_stream(messages:list,tools:listNone):body{model:mimo-v2-flash,messages:messages,stream:True,# 关键开启流式}iftools:body[tools]tools# streamTrue 告诉 requests 不要一次读完responserequests.post(url,jsonbody,streamTrue)forlineinresponse.iter_lines():lineline.decode(utf-8)ifline.startswith(data: ):dataline[6:]ifdata[DONE]:breakchunkjson.loads(data)contentchunk[choices][0][delta].get(content,)ifcontent:yieldcontent和 Day 1 的区别Day 1:response requests.post(...)→ 等完整响应Day 8:streamTrueiter_lines()→ 边收边显示五、流式 工具调用核心难点这是今天最难的部分。普通流式只需要拼接文本但加上工具调用就复杂了普通流式 chunk1你 chunk2好 → 直接拼显示 带工具调用的流式 chunk1让我查一下时间 chunk2{tool_call: get_current_time} ← 需要停下来 → 执行工具拿到结果 → 把结果发回 API继续流式 chunk3现在是 16:075.1 MiMo API 的工具调用格式经过调试发现MiMo API 的 tool_call 是完整一个 chunk发过来的// 第1个 chunkAI 开始说话{choices:[{delta:{content:我来帮你查看},finish_reason:null}]}// 第2个 chunk工具调用完整的一个 chunk不是分多个{choices:[{delta:{tool_calls:[{index:0,id:call_xxx,function:{name:get_current_time,arguments:{}}}]},finish_reason:null}]}// 第3个 chunk结束标记{choices:[{delta:{},finish_reason:tool_calls}]}5.2 核心代码defstream_with_tools(messages,tools,max_rounds3):forround_numinrange(max_rounds):# 调用 APIresponserequests.post(url,jsonbody,streamTrue)text_contenttool_calls_map{}has_tool_callsFalseforlineinresponse.iter_lines():# 解析 chunk...ifcontent:yield{type:text,content:content}iftool_calls:has_tool_callsTrue# 收集工具调用信息# 如果有工具调用ifhas_tool_callsandfinish_reasontool_calls:# 执行工具fortcintool_calls_map.values():resultexecute_tool(tc[name],tc[arguments])yield{type:tool_call,name:tc[name]}yield{type:tool_result,result:result}# 工具结果加入消息历史continue# 继续下一轮else:return# 正常结束关键逻辑检测到 tool_call → 执行工具 → 结果发回 API → 继续流式。最多循环 3 轮。实测效果文字 → 我来帮你查看当前时间。 工具 → get_current_time() → 2026年06月07日 16:07:10 文字 → 现在是 **2026年6月7日星期六16:07:10**。六、SQLite 持久化 会话管理6.1 数据库设计作为 Java 后端看到这个设计秒懂-- 会话表CREATETABLEsessions(idINTEGERPRIMARYKEYAUTOINCREMENT,titleTEXTNOTNULLDEFAULT新对话,created_atTEXTNOTNULL,updated_atTEXTNOTNULL);-- 消息表CREATETABLEmessages(idINTEGERPRIMARYKEYAUTOINCREMENT,session_idINTEGERNOTNULL,roleTEXTNOTNULL,-- user / assistant / toolcontentTEXT,tool_callsTEXT,-- JSON 格式的工具调用tool_call_idTEXT,created_atTEXTNOTNULL,FOREIGNKEY(session_id)REFERENCESsessions(id));Java 对照Entity class Session → sessions 表 Repository → ChatDatabase 类 Service ChatService → PersistentChat 类 RestController → FastAPI app.post SseEmitter → StreamingResponse JdbcTemplate → sqlite3.connect()6.2 踩坑SQLite 连接冲突写save_message的时候踩了个坑defsave_message(self,...):withsqlite3.connect(self.db_path)asconn:conn.execute(INSERT INTO messages ...)self.touch_session(session_id)# ← 这里又开了一个连接conn.commit()save_message开了一个连接touch_session又开一个两个连接同时写同一个数据库 →死锁修复内联更新一个连接搞定。defsave_message(self,...):withsqlite3.connect(self.db_path)asconn:conn.execute(INSERT INTO messages ...)conn.execute(UPDATE sessions SET updated_at ? ...)# 内联conn.commit()教训Java 里用连接池自动管理Python 的 sqlite3 要手动注意连接生命周期。七、完整 Web 应用最后一步把所有东西组装成一个 Web 应用app.post(/api/sessions/{session_id}/chat)defchat(session_id:int,req:ChatRequest):defevent_generator():foreventinchat_service.chat(session_id,req.message):yieldfdata:{json.dumps(event)}\n\nyielddata: [DONE]\n\nreturnStreamingResponse(event_generator(),media_typetext/event-stream)前端用原生 HTML/JS核心是fetchReadableStream读取 SSEconstresponseawaitfetch(/api/sessions/1/chat,{method:POST,body:JSON.stringify({message:msg})});constreaderresponse.body.getReader();while(true){const{done,value}awaitreader.read();if(done)break;// 解析 SSE 事件实时更新页面}功能清单✅ ChatGPT 风格的聊天界面✅ 流式打字效果✅ 多会话管理新建、切换、删除✅ 对话持久化SQLite关了再开还在✅ 工具调用时间查询、计算器八、文件依赖关系shared_config.py ← 全局 API 配置API Key、模型 ↓ 05_streaming_tool_agent.py ← 流式 工具调用核心 ↓ 06_persistent_chat.py ← 持久化 会话管理 ↓ 07_chat_web_app.py ← Web 应用复用上面两个九、Day 8 总结9.1 今日收获Day 8 流式输出 工具调用 持久化 Web 应用 yield 生成器Python 版 Iterator但简洁 10 倍 SSE 协议 服务器推送 Java 的 SseEmitter 流式 API requests(streamTrue) iter_lines() 工具调用 流式中检测 tool_call → 执行 → 继续 SQLite Python 版 JDBC手动注意连接冲突 Web 应用 FastAPI 原生 HTML 迷你豆包9.2 最大的感受今天是从会用到能做东西的转折点。前7天学的都是零散的知识点API调用、Agent概念、RAG、LangChain…今天第一次把它们组装成了一个完整的、能跑的、有界面的应用。虽然 UI 简陋虽然功能简单但核心架构和 ChatGPT、豆包是一样的流式输出 ✅工具调用 ✅多轮对话 ✅持久化 ✅Web 界面 ✅一个 Java 后端用 Python 手撸了一个 AI 聊天应用这感觉还挺爽的。系列进度Day 8 / 90学习节奏周日大肝从 yield 到完整 Web 应用下一阶段接入更多工具、优化 UI、部署上线