从零构建本地优先AI Agent：Ktor + Ollama 流式交互架构实践

我一直想找一个本地优先的 AI Agent 来帮我处理日常的文本杂活试过几个现成工具要么太重要么功能核心锁在云端后来索性自己动手用 Ktor 从零写了一个开源的方案。所谓本地优先我的理解很直接模型推理可以走本地 Ollama工具调用在本地执行会话记录落在自己的 SQLite 文件里一切不以某个远端服务为核心。Ktor 在这里的角色是把整个 Agent 的交互能力暴露成 HTTP 接口打开浏览器就能聊也可以被其他程序当作普通 API 调用。这篇文章不是讲什么惊艳成果更多的是一份项目复盘。我会把为什么选 Ktor、整体架构怎么拆、哪些代码值得直接抄以及实际踩过的几个坑都摊开说。适合正在用本地模型、想自己折腾一套 Agent 框架的开发者也适合对 Kotlin/JVM 服务端感兴趣、想看看 Ktor 在处理流式接口和长任务调度时到底怎么用的朋友。1. 为什么坚持本地优先又为什么是 Ktor1.1 本地优先不是情怀是三个现实需求先说隐私。把聊天记录和文档丢给云端服务等于默认信任对方的隐私政策。但是我本地跑的模型对话内容、分析结果、工具产生的临时文件全部留在我的磁盘里没有第三台服务器经过。对于写技术方案、整理个人事务这类内容这个安心感很重要。再说成本。云端 API 按 token 计费长上下文对话一旦展开开销会快速上涨。本地跑模型更像买硬件固定成本虽然一次性投入不小但用多了边际成本几乎为零。而且本地推理没有并发限制我想开几个会话就开几个不用担心厂商的配额策略。最后是可控性。云端 Agent 通常给你一个黑盒内部怎么调模型、怎么解析工具参数用户完全看不到。本地优先意味着所有环节都可观测、可修改、可回放。我遇到一次异常可以把当时的模型输入输出全部捞出来分析这种调试粒度是云端方案给不了的。1.2 FastAPI、Spring Boot 都不差但 Ktor 更贴合做服务端方案摆在我面前的无非是 Python 系 FastAPI、Java/Kotlin 系 Spring Boot以及 Ktor。我最终选了 Ktor核心原因是它把轻量、协程原生、类型安全这三件事平衡得最好。FastAPI 写原型确实快但项目一旦涉及工具注册、任务调度、本地文件索引这类偏底层工程的部分Python 的动态类型会让维护成本慢慢变高。Spring Boot 功能全面可对于我这个项目来说太重了启动几百毫秒不说依赖和配置光是看着就头大。Ktor 是 Kotlin 协程的原生公民处理高并发长连接非常自然。我用它既能写服务端也能用同一套 Ktor Client 去请求本地 Ollama整个项目都是 Kotlin类型定义和序列化配置可以前后端共享这一点是跨语言组合很难替代的。对比项KtorSpring BootFastAPI启动速度快较慢快流式响应支持原生协程 SSE需 WebFlux 或额外配置异步支持好类型安全强强弱依赖体积小大中与 Kotlin 生态契合极佳尚可无关1.3 开源的意义和真实代价开源这个动作本身不复杂但后续维护是实打实的工作。我选择把项目开源一是因为本地 Agent 这类工具本来就带有很强的个人色彩不同人的使用习惯差异极大社区反馈能帮我把通用性打磨得更好。二是开源等于强制我写清楚文档和示例否则别人根本跑不起来这种约束反过来能提升代码质量。不过要有心理准备开源之后你会收到各种 issue有的确实是 bug有的只是配置环境不同。我的经验是项目初期别追求功能多先把核心链路稳定住再慢慢吸收外部需求。许可证方面我选了 MIT对使用者最宽松社区也最容易接受。2. 整体架构与各模块分工2.1 分层设计从 HTTP 入口到模型推理这个项目没有采用花哨的微服务架构就是一个单体 JVM 进程内部按职责拆成四层。最外层是 Ktor 路由层负责接收 HTTP/SSE 请求、鉴权和参数校验往里走是 Runner 层也就是 Agent 的主循环它管理每轮对话的推理、工具调度和上下文拼接再往里是工具层每个工具就是一个可插拔的执行单元最底层是存储层负责对话记录、会话状态和本地文档索引的读写。这样拆分的好处是每一层替换起来都很容易。我今天接 Ollama明天如果换一个支持 OpenAI 协议的推理服务只需要改 Runner 层里面的 client 适配路由和存储完全不动。同理想加新的工具能力不用碰 Agent 循环只要在工具注册表里多填一条记录。2.2 模型接入层用 Ktor Client 对接 Ollama本地推理我用的是 Ollama 作为模型运行时它对外提供 HTTP API想直接用curl请求都行。但在 Ktor 里我选择用 Ktor Client 来管理这些请求这样可以复用已有的协程调度、超时控制、连接池配置不用再引一套 HTTP 库。Ollama 的流式接口返回的是 NDJSON 格式每一行都是一个 JSON 对象。做流式响应时不能等整个响应体读完再解析那样首字延迟会高得离谱。正确的做法是按行读取输入流每读到一个合法 JSON 行就立刻解析、立刻推给前端。这里有个容易忽略的点Ktor Client 默认可能对响应体做整体缓冲需要在请求级别关闭或者用 IO 流方式读取。我用的是 prompt-based 工具调用方案而不是模型原生 function calling。原因很简单本地小模型的工具调用能力参差不齐直接走原生的 function calling 格式经常出现参数名对不上、JSON 格式漂移的问题。prompt 方案是让模型输出格式化的文本然后用一个独立解析器把文本转成结构化工具调用。灵活度更高换模型也不用改逻辑。2.3 工具注册机制把工具当成接口来设计工具层是整个 Agent 的执行末梢。我设计了一个类似路由注册的机制每个工具提供名字、描述、参数定义和执行函数四要素然后统一注册到一个工具表里。描述信息很关键因为本地模型要根据描述来决定是否调用工具以及传什么参数。如果只写“获取天气”模型很难知道这个工具有没有用如果写成“获取指定城市未来三天天气预报城市名用中文”模型调用准确率会明显提升。参数定义我用的是 JSON Schema 格式解析工具入参的时候直接用 Kotlin 序列化库把 JSON 转成 typed 对象省去一堆手写解析代码。工具执行时的权限也要收紧。我的做法是在执行函数外层套一层校验器读文件路径必须限定在根目录内执行本地命令必须走命令白名单避免模型被 prompt injection 之后做出危险操作。这点后面会细聊但属于工具链设计里最容易忽略的部分。2.4 存储层SQLite 加文件目录双轨制对话记录和会话状态我用 SQLite 存结构清晰、查询方便单机场景完全扛得住。文档和工具输出则直接用文件目录存好处是人眼可以直接查看也可以随时丢进任何文本处理管道。一开始我尝试全用 JSON 文件存储对话后来发现会话多了之后按时间线检索和多会话汇总非常吃力。换成 SQLite 之后会话、消息、工具调用记录都可以用 SQL 做聚合排查问题效率高得多。文件目录负责非结构化数据比如知识库文档切片、Agent 生成的临时报告这样两类数据用各自最擅长的方式保存。SQLite 有一个高频踩坑点是并发写入后面在问题排查章节我会专门讲怎么用 WAL 模式解决。3. 核心代码与关键实现细节3.1 用 SSE 把 Agent 思考过程推给前端Ktor 写普通 JSON 接口很简单但要实时推送 Agent 的思考过程最省事的方案是 SSEServer-Sent Events。它基于 HTTP不需要 WebSocket 那么重的连接管理浏览器原生支持断线重连也自动处理好了。下面这段路由实现了 SSE 流向客户端推送事件fun Application.configureRouting() { routing { route(/api/agent) { get(/stream) { val sessionId call.request.queryParameters[session] ?: default call.respondTextWriter(contentType ContentType.Text.EventStream) { val flow agentEngine.stream(sessionId) flow.collect { event - when (event) { is AgentEvent.Start - write(event: start\ndata: ${event.sessionId}\n\n) is AgentEvent.Token - write(data: ${event.text}\n\n) is AgentEvent.ToolStart - write(event: tool_start\ndata: ${event.toolName}\n\n) is AgentEvent.ToolResult - write(event: tool_result\ndata: ${event.output}\n\n) is AgentEvent.Done - write(event: done\ndata: ${event.costMs}\n\n) } flush() } } } } } }SSE 的格式要求很严格每个事件用两个换行符结尾event:字段规定事件类型data:字段放内容。浏览器端的 EventSource 会自动按这个格式解析。这里必须调flush()否则数据会积压在缓冲区里前端迟迟收不到消息。我用 Kotlin 的 Flow 作为事件的统一载体定义了一个 sealed interface 把所有事件类型收拢在一起分支处理时用when穷尽编译器会兜底检查少写分支都过不了编译。3.2 Agent 主循环流式推理与工具调用的编排Agent 的核心逻辑是一个典型的循环把用户输入拼到消息列表里请求模型生成回复检查回复里有没有工具调用意图有就执行工具并把结果追加进上下文没有就结束这轮对话。这里的关键是每一轮循环都要保证上下文不失控既要携带工具执行结果又不能把历史日志无脑塞进去。suspend fun run(userInput: String) { messages ChatMessage(user, userInput) while (true) { val reply withTimeout(90.seconds) { llm.chat(messages, stream false) } val toolCalls ToolCallParser.parse(reply) if (toolCalls.isEmpty()) { messages ChatMessage(assistant, reply) eventSink.send(AgentEvent.Token(reply)) break } for (toolCall in toolCalls) { eventSink.send(AgentEvent.ToolStart(toolCall.name)) val result toolRegistry.execute(toolCall.name, toolCall.arguments) eventSink.send(AgentEvent.ToolResult(result)) messages ChatMessage(tool, result) } } eventSink.send(AgentEvent.Done(costMs)) }我给模型的单轮推理设置了总超时防止模型一直生成停不下来。生产环境实测下来90 秒对于本地 7B 模型已经非常宽裕如果超过这个时间多半是模型陷入重复输出死循环而不是良性的长思考。超时抛出异常后要把当前会话标记为中断态避免残留的上下文继续干扰下一轮对话。注意withTimeout抛出的是TimeoutCancellationException它属于CancellationException的子类。如果你在更外层用runCatching捕获所有异常会把这个中断信号吃掉导致协程无法真正取消。这是协程编程里很经典的一个暗坑必须单独处理超时异常或直接让它往外抛。3.3 会话状态管理每个会话一个协程作用域并发聊天场景下状态管理的核心诉求是隔离。我做了一张会话表每个会话对应一个独立的CoroutineScope内部用SupervisorJob让子任务失败不影响整个会话生命周期。class SessionManager { private val scopes ConcurrentHashMapString, CoroutineScope() fun scopeOf(sessionId: String): CoroutineScope { return scopes.computeIfAbsent(sessionId) { CoroutineScope(SupervisorJob() Dispatchers.Default) } } fun dispose(sessionId: String) { scopes.remove(sessionId)?.cancel() } }模型推理虽然是 IO 密集操作但本地模型的 tokenize、prompt 拼接和工具参数解析都有不少 CPU 计算放Dispatchers.Default能更好利用多核。不过要注意如果同时开太多会话默认线程池会被打满影响整个 JVM 的响应。我在实际使用中给Parallelism设了上限也就是同一时间最多跑 4 个 Agent 循环其余请求排队等待。3.4 本地 RAG 的轻量实现不引入重向量库知识库问答是这个 Agent 的一个重要场景。我一开始想接向量数据库后来果断放弃。本地小模型本身语义能力有限文档集规模通常也就几千个切片用 BM25 关键词打分加简单的文本分块效果并不差而且省掉了嵌入模型的额外显存开销。检索功能我实现得尽量朴素。文档切块按固定长度加重叠窗口切切出来的片段写到本地文件目录文件名带文档 ID 和切片序号。查询时逐目录扫描用 BM25 分数排序取 Top-K 片段拼进 prompt。data class Chunk(val docId: String, val index: Int, val text: String) fun retrieve(query: String, topK: Int 5): ListChunk { val tokenizer SimpleTokenizer() return Files.walk(dataDir) .filter { it.toString().endsWith(.json) } .map { readChunk(it) } .map { chunk - chunk to score(query, chunk.text, tokenizer) } .sortedByDescending { it.second } .take(topK) .map { it.first } .toList() }这种朴素方案在文档数量上千时完全可用但有几个明显的坑。中文分词如果只按单个汉字切会丢掉大量语义信息我用了简单二元组切分把相邻两个字作为一个索引单位对中文检索效果提升很明显。另外切片重叠太密会让同一个内容反复出现在多个切片里既浪费 token 又影响去重重叠率控制在 10% 到 20% 之间比较合理。4. 从 0 到 1 复现这个项目的实操记录4.1 环境准备与 Gradle 依赖配置复现前先把环境准备好。我用 JDK 21Gradle 走 Kotlin DSLKtor 使用 3.x 版本。核心依赖清单如下plugins { kotlin(jvm) version 2.0.20 kotlin(plugin.serialization) version 2.0.20 id(io.ktor.plugin) version 3.0.0 } dependencies { implementation(io.ktor:ktor-server-core) implementation(io.ktor:ktor-server-netty) implementation(io.ktor:ktor-server-content-negotiation) implementation(io.ktor:ktor-serialization-kotlinx-json) implementation(io.ktor:ktor-client-core) implementation(io.ktor:ktor-client-cio) implementation(org.xerial:sqlite-jdbc:3.46.1.3) implementation(ch.qos.logback:logback-classic:1.5.8) }Gradle 版本用 8.10 以上就好。Ktor 插件会帮你统一 BOM 版本管理Ktor 各模块之间不需要逐个手动对齐版本号这是它相比手写一堆依赖的一个省心点。4.2 项目目录结构我按功能分包而不是按技术分层分包这样每加一个新能力改动范围都集中在一个包内不容易破坏全局结构。src/main/kotlin/com/example/agent/ ├── Application.kt # Ktor 入口加载配置和路由 ├── model/ │ ├── ChatMessage.kt # 消息体定义 │ └── AgentEvent.kt # SSE 事件定义 ├── agent/ │ ├── AgentEngine.kt # 主循环 │ ├── ToolCallParser.kt # 工具调用解析 │ └── SessionManager.kt # 会话作用域管理 ├── tools/ │ ├── ToolRegistry.kt # 工具注册表 │ └── builtin/ # 内置工具实现 ├── store/ │ ├── SqliteStore.kt # SQLite 读写 │ └── ChunkStore.kt # 文档切片存储 └── llm/ └── OllamaClient.kt # Ollama HTTP 客户端启动流程很直接Application.kt里做三件事加载配置文件和初始化存储层构建 AgentEngine 和 ToolRegistry最后注册路由。因为用了懒加载的by lazy初始化大对象冷启动时间可以控制在 1 秒以内。4.3 一次完整请求的链路追踪我们走一遍标准请求链路帮助你理解各部分如何协作。客户端发来一条消息比如“请帮我总结 docs 目录下关于 Ktor 的文章”。Ktor 路由层先解析参数把 sessionId 和消息内容提取出来然后交给 AgentEngine。AgentEngine 先从 SessionManager 拿会话作用域和消息历史再把新消息追加进列表。Runner 调用 OllamaClient 发请求模型返回一段包含工具调用意图的文本ToolCallParser 解析出search_docs工具和参数{keyword: Ktor}。接着 ToolRegistry 执行搜索工具扫描 docs 目录用前面说的朴素检索逻辑返回 Top-5 文档摘要。这些摘要被追加进消息列表再次请求模型。这次模型拿到工具结果后生成一篇完整总结不再触发工具调用数据被持久化到 SQLite。日志输出大概是这样的[INFO] 10:00:01.234 [sessionIda1b2] receive user message [INFO] 10:00:01.512 [sessionIda1b2] call model, max_tokens2048 [INFO] 10:00:05.103 [sessionIda1b2] parse tool call: search_docs(keywordKtor) [INFO] 10:00:05.211 [sessionIda1b2] execute tool: search_docs, cost86ms [INFO] 10:00:06.450 [sessionIda1b2] call model again, context tokens4832 [INFO] 10:00:09.872 [sessionIda1b2] conversation done, total cost8734ms这条链路走完大概 8 到 9 秒其中大头在本地模型两次推理。搜索工具本身只花了 86 毫秒说明 RAG 检索的耗时和模型推理不在一个量级优化重点应该放在推理环节而不是检索环节。4.4 发布到 GitHub 的开源基本动作开源发布这件事做成流水线之后其实很省事。我推荐在 GitHub 上建仓后先补上三样东西LICENSE 文件、README、示例配置。README 里必须写清楚快速启动命令、依赖要求和默认端口否则别人 clone 下来根本跑不起来。CI 我用 GitHub Actions 配了两条流水线一条跑./gradlew build和单元测试另一条负责发布二进制包。JVM 项目发布相对简单构建产物是一个 fat JAR直接加个application插件就能生成可执行脚本。配好 release 流水线之后每次打 tag 就会自动构建并上传资产省去手动上传的麻烦。版本号从 0.1.0 开始就行不必一上来就搞 1.0。开源项目最重要的是持续迭代版本号的意义在于标记行为变化早期版本频繁变化很正常。5. 常见问题与排查实录5.1 模型卡住不动超时、取消与线程池的关系我在实际使用里遇到最多的一个问题是前端连上 SSE 后明明模型还在生成界面却像冻住了一样。排查后发现原因是 Ktor 路由的请求作用域被取消了导致 Runner 里的协程被连带取消但模型那侧的底层连接没有主动关闭就像拔了网线但是电话没挂。解决思路是区分两种语义。短对话场景用户关闭页面时取消请求是合理的协程取消还能帮我们及时释放资源。但长任务场景比如模型正在执行一个耗时的工具调用不应该跟着 HTTP 连接一起取消。我给这类长任务专门建了一个非请求作用域的协程任务的执行状态落在数据库里前端重新连接后可以从任务表中恢复进度。另外要注意线程池配置。Ktor Netty 的默认事件循环线程是 CPU 核心数相关的如果你在事件循环线程里跑了阻塞的 SQLite 查询整个服务的吞吐会瞬间掉下来。我的做法是所有可能阻塞的操作都用withContext(Dispatchers.IO)包裹避免污染 Netty 的线程池。5.2 本地模型首字延迟高聊天像在等便秘本地模型跑起来之后有个现象前几轮对话还好越聊越慢甚至第一句话要等好几秒。核心原因是模型进程在空闲一段时间后会被 Ollama 卸载需要重新加载权重。我实测了一个 7B 模型的数据状态首字延迟说明冷启动加载2.5s ~ 4s模型重新载入显存热加载空闲 10 分钟1.2s ~ 2s部分权重被换出持续使用中0.5s ~ 1s权重常驻显存解决的思路有两个层面。一是调整 Ollama 的keep_alive参数把模型常在内存时间设长一点比如30m对于个人机器够用了。二是做预热请求应用启动后立刻发一个“你好”的请求强制模型加载。两个办法叠加之后用户感知到的首字延迟基本稳定在 1 秒以内。还有一个容易忽略的点是模型上下文长度。本地模型如果上下文设得太长比如 32K显存会被上下文占掉一大块推理速度会明显下降。我默认设 8K足够覆盖日常对话和工具调用场景。如果确实需要长文档分析就单独建一个长上下文会话不要把默认会话的历史无限拉长。5.3 SQLite 锁冲突与并发写入SQLite 单机确实够用但并发写入是个坑。多个会话同时落对话记录时偶发database is locked错误。这是因为 SQLite 默认的 rollback journal 模式在写操作时会对整个库加锁并发写会互相阻塞甚至超时。解决办法是开启 WAL 模式。WAL 全称 Write-Ahead Logging写操作先追加到日志文件不直接阻塞读读写并发能力会好很多。我每次初始化连接时执行这条 PRAGMAPRAGMA journal_modeWAL; PRAGMA synchronousNORMAL;其中synchronousNORMAL在 WAL 模式下能保持数据安全性的同时降低写入延迟。另外写成单一写连接比多连接更省心。我封装了一个数据访问层强制所有写操作走同一个连接读操作可以走独立连接这样能彻底避开 SQLite 的锁竞争。5.4 Kotlin 序列化与动态 JSON 的坑工具调用参数本质是动态 JSON你事先不知道工具会收到什么字段这跟 Kotlin 强类型风格是矛盾的。我最初用data class反射解析所有参数结果工具一多就乱套新增字段还要改类代码。后来我把工具参数定义统一成JsonObject类型在执行工具函数内部再手动解析具体字段。这样注册工具时不需要定义一套对应的 Kotlin class扩展工具的流程大大简化。代价是失去了一部分编译期类型检查所以我在每个工具函数里加了参数校验逻辑缺字段直接报工具级错误不会让错误蔓延到 Agent 主循环。还有一个坑是 Kotlin 的Json.encodeToString对原生Long类型处理。前端 JS 的数字精度不够超过安全整数范围的值会被截断。我在配置文件里开启了JsonNumber作为长整数字面量并且给序列化器设置ignoreUnknownKeys true避免模型返回多余字段时反序列化报错。5.5 静态资源与 API 路由冲突以及环境切换的问题项目带了简单的前端页面Ktor 用staticResources挂载静态资源。一开始我把静态路由挂在根路径/API 路由挂在/api下自以为没问题。结果前端页面刷新时Ktor 会把请求匹配到前端路由返回 404。原因是 Ktor 路由是按注册顺序匹配的静态资源处理器注册太早把所有路径都揽过去了。解决方法是把静态资源挂到/web路径下用 Nginx 或者其他网关做路径转发。开发环境则直接用http://localhost:8080/web/访问简单粗暴。多环境配置也值得一提。本地开发和打包 prod 模式时Ollama 地址可能不同我用 Ktor 的application.yaml加环境变量占位符默认值指向http://127.0.0.1:11434部署到别的机器时通过环境变量覆盖。问题常见原因处理手段SSE 前端收不到消息忘记 flush每条事件写入后立即 flush会话越聊越慢上下文无限累积设置 8K 上下文上限模型返回工具调用失败prompt 工具样例不够每个工具附 1~2 个调用示例SQLite 文件锁多连接并发写开启 WAL单一写连接页面刷新 404静态路由与前端路由冲突拆分路径前缀我自己的使用习惯是把 Agent 每次对话和工具调用的原始日志都写到一个可回放的文件里本地优先的好处这时候就体现出来了。线上服务出问题只能靠日志猜本地 Agent 出问题我可以直接重放整段对话连模型当时的原始输出都看得一清二楚。这种可调试性是我做完这个项目之后最不想放弃的部分。工具链、模型和存储层都换过不止一次唯独这个回放日志的习惯保留到了最后也推荐给所有想在本地跑 Agent 的朋友。