尧图网页设计 
1. 为什么我会自己折腾一个智能体运行时框架先交代一下背景。我在团队里主要负责把大模型能力落进实际业务差不多把市面上能见到的智能体平台和开源框架都摸过一圈之后发现一个尴尬的现状平台型的智能体搭建工具确实上手快但一旦涉及复杂的业务交互比如多轮状态保持、权限差异、动态工具路由、流式回包拼接平台那套可视化节点就开始力不从心而主流开源智能体框架功能全但重量也不小依赖一堆、抽象层层叠叠调试一个问题要翻好几层源码。我的需求其实很朴素要一个能写进现有服务里的智能体运行时不挑部署环境启动够快逻辑透明复杂交互我能自由编排。市面上没有完全合我胃口的于是就有了 Flowing 这个项目。它不是一个搭建平台而是一个运行时框架——你写代码来定义智能体的行为框架负责把大模型调用、工具执行、状态流转、多轮交互这些繁琐的底层逻辑收拢起来。这篇文章就把我设计 Flowing 时的核心思路、实际代码示例和踩过的坑都摊开讲如果你也有类似诉求可以参考着做出自己的选择。2. Flowing 的设计取舍轻量级不等于功能少而是去掉你不需要的部分轻量级这几个字被用滥了很多框架说自己轻量实际装完依赖一看体积比业务代码还大。我给 Flowing 定的标准很具体核心运行时没有任何必须的第三方大依赖只依赖标准库和一个轻量的 LLM 客户端接口定义。其他能力比如具体的模型厂商 SDK、向量存储、消息队列全部通过插件方式按需挂载。这样做的直接好处是你在一个只有几百 MB 内存的容器里也能跑起来而且每一次启动几乎是瞬时的没有一堆初始化检查。面向复杂交互这个定位决定了不能照着聊天机器人那套模式做。我拆解了常见的智能体交互场景抽象出三个核心原语状态State、事件Event、步骤Step。状态就是当前会话的上下文包括用户输入、中间变量、历史记录事件是外部触发的信号比如用户消息、定时任务、Webhook 回调步骤是智能体处理事件的最小单元可以是一个大模型调用、一段工具代码、一个条件判断也可以是另一个子智能体。框架的本质是一个按顺序执行步骤、并根据步骤结果更新状态的事件循环。这样做对比起图编排或链式调用的好处是你可以很容易地表达循环、嵌套和动态分支而这些正是复杂交互的刚需。很多人会问直接用 async 循环自己写不就行了吗确实可以但遇到重试、超时、上下文裁剪、并行执行几个子任务再汇总这些逻辑散落在业务代码里会非常啰嗦。Flowing 把这些横切关注点内聚到运行时里你的业务代码只需要描述做什么而怎么调度、怎么容错由框架兜底。3. 半小时跑通第一个 Flowing 智能体3.1 安装与项目结构Flowing 是一个 Python 包Python 版本要求 3.10 以上这主要是为了用上最新的类型语法和match语句。安装很简单pip install flowing-agent装完之后推荐按下面的目录结构组织你的项目这不算强制但我试过很多种布局后觉得这个最顺手my_agent/ ├── agent.py # 定义智能体的入口和步骤链 ├── steps/ # 业务相关的自定义步骤 │ ├── __init__.py │ └── fetch_order.py ├── tools/ # 智能体可调用的外部工具 │ └── query_db.py └── config.yaml # 模型、超时、日志等配置3.2 定义状态模型和第一个步骤第一步是定义状态。Flowing 内置了一个BaseState类你只需要声明字段运行时负责序列化和传递。为了支持复杂交互状态本身是可变对象每个步骤执行后都可以直接修改它。from flowing import BaseState class ChatState(BaseState): user_input: str user_id: str history: list[dict] intent: str | None None result: str 接着写一个步骤。步骤是实现Step接口的类关键是async def run(self, ctx)方法。ctx里能拿到当前状态、运行时上下文、以及一个markdown_text之类的辅助函数。我习惯把提示词模板也放在步骤类里这样做的好处是每个步骤自包含测试起来很方便。from flowing import Step, Context class UnderstandIntent(Step): async def run(self, ctx: Context[ChatState]): messages [ {role: system, content: 你是一个意图识别器只输出一个词查询、或执行。}, {role: user, content: ctx.state.user_input}, ] resp await ctx.llm.chat(messages) ctx.state.intent resp.text().strip() return self.OK3.3 组装智能体并启动服务Flowing 的智能体对象由一系列步骤组成运行时会按顺序执行。但不是死板的顺序步骤里可以通过返回状态码来提前中断或跳转这个机制后面会详细说。先看一个简单例子from flowing import Runtime, Step agent Runtime[ChatState]( stateChatState, steps[ UnderstandIntent(), SelectBranch(), CallToolOrReply(), ], # 这里传入你要接入的 LLM 客户端支持流式 ) if __name__ __main__: agent.serve(port8000)这样就得到了一个 HTTP 服务支持POST /chat接口请求体里{user_input: 帮我查一下昨天的订单, user_id: u_123}返回流式响应。整个过程没写任何 FastAPI 代码运行时直接把常用协议封好了。如果你不想起服务也可以在脚本里直接调用await agent.run_state(state)适合做单元测试或者批处理。4. 复杂交互的核心分支、循环、工具调用与子智能体这一章是 Flowing 真正区别于用 Python 手写代码的地方。我遇到过很多所谓智能体框架只能做线性的调一次 LLM - 执行一个工具的流程稍微复杂一点的业务就捉襟见肘。我设计 Flowing 时专门针对以下几个交互模式做了支持。4.1 带条件的动态路由业务里经常要按意图走不通的分支比如识别到查询就走检索步骤识别到下单就走校验步骤。Flowing 用步骤的next显式指定来支持路由同时允许条件表达式class SelectBranch(Step): async def run(self, ctx): if ctx.state.intent 查询: return self.GOTO(search_order) elif ctx.state.intent 执行: return self.GOTO(execute) return self.FAILGOTO的目标是步骤的名字运行时维护一个步骤注册表。你可以在Runtime初始化时命名步骤也可以让步骤类自己声明名字。这样流程就不是一条直线而是一张跳转表跳出循环和复杂状态机都方便。4.2 工具声明与自动调用智能体绕不开工具调用。Flowing 提供了一种简洁的工具注册方式不需要你手动解析大模型返回的 JSON 字符串from flowing import tool tool(description根据订单ID查询订单详情, params{order_id: string}) async def fetch_order(order_id: str) - dict: # 里面可以是任何查库、调外部 HTTP 的逻辑 result await query_db(order_id) return {order_id: order_id, status: result.status}在步骤里你可以显式调用这个工具也可以让大模型自主决定调用class ToolOrReply(Step): async def run(self, ctx): if ctx.state.intent 查询: tool_result await ctx.call_tool(fetch_order, order_id12345) ctx.state.result tool_result else: messages [{role: user, content: ctx.state.user_input}] ctx.state.result await ctx.llm.chat(messages) return self.OK自主调用模式需要 LLM 供应商支持 function calling 协议Flowing 的模型客户端接口里内置了工具描述传参你只需要在创建 LLM 客户端时传入工具列表。框架运行时会解析模型的函数调用请求自动执行工具并回填结果不用你手动拼接多轮回调消息。这一块当然需要与具体模型调接口但接口已经抽象得比较干净了像我实际用 OpenAI 兼容接口和通义千问的兼容接口都是改一行 base_url 就行。4.3 并行执行与结果合并复杂交互里经常要同时查多个数据源再汇总回答。Flowing 提供了一个FanOut步骤类型接收子步骤列表并行执行from flowing import FanOut class ParallelSearch(FanOut): steps [search_orders, search_products] async def aggregate(self, results): ctx.state.aggregated | .join(results) return self.OKaggregate在子步骤全部完成后被调用。并行执行用的是 asyncio 的gather所以你的子步骤里不能有阻塞式 IO如果有记得改成await asyncio.to_thread()。我踩过这个坑一开始在工具里直接用了requests结果并行直接变成了串行吞吐惨不忍睹后来统一改成httpx.AsyncClient才解决。4.4 子智能体把一个智能体看作一个步骤单个智能体管不了太巨大的任务Flowing 允许你在一个步骤里启动另一个智能体实例子智能体有自己的状态、步骤和模型配置。这样做在多角色协作场景里特别好用主智能体做任务拆解把具体环节交给搜索智能体、客服智能体执行再把结果汇总回主状态。实现上子智能体的 Runtime 实例就是一个普通的 Step因为它也需要输入输出。关键在于子智能体的状态隔离子智能体的状态是独立对象不能直接碰父智能体的内部字段只能通过输入参数传入和返回值传递结果。这个约束一开始我觉得麻烦但实际用下来避免了很多隐式耦合出了 bug 也好定位。5. 实际运行中的性能表现与避坑记录5.1 和常见框架的对比我拿我自己之前用过的典型平台和框架做了个不严谨的对比测试场景是 20 个并发用户、每个用户连续 10 轮多轮交互夹杂工具调用记录 P95 响应时长和框架初始化耗时从启动到可服务方案初始化耗时P95 响应依赖体积交互编排能力Flowing180ms2.1s核心约 1MB高原生支持跳转/循环/子智能体某全流程框架3.2s2.4s约 80MB中支持链式循环需要 hack某个平台依赖外部托管2.8s不适用中可视化编排但无法做复杂代码逻辑这个表格不是要说明 Flowing 比谁强而是想说对于已经熟悉代码、且业务逻辑有大量定制需求的团队一个轻量运行时带来的直接收益是启动快、好调试、依赖少。平台的定位是让不懂代码的人也能搭框架是让懂代码的人不受约束。两者的设计目标不同。5.2 踩过的三个坑状态序列化的问题。复杂状态里如果包含自定义对象保存会话快照时会挂。我后面统一规定状态字段只允许 JSON 可序列化类型自定义对象必须放步骤内部属性。这个规定虽然限制了一些自由度但换来了能方便地把状态存入 Redis 做持久化。流式输出的拼接与中断。大模型流式返回文本片段如果中途用户发来新消息需要打断当前响应我最初的处理是直接拿到当前 token 终止结果下一次恢复上下文时缺了半截话。后来我设计成取消当前步骤但把已经生成的 token 记录在状态里下一次响应先把这部分列出再开始新回复。稍微有点绕但用户体验好很多。工具调用的递归深度。某些场景下模型会连续调用十几个工具如果工具本身又触发新的子工具可能会无限递归。Flowing 里给每个运行实例加了一个max_tool_rounds参数默认 10超过直接报错防止把 token 烧光。5.3 什么时候用 Flowing什么时候不要用如果你只是做一个展示用的问答机器人直接调 LLM API 就够了不需要框架如果你想在一个大型平台里做复杂的业务流程且你团队里有能写 Python 的工程师那 Flowing 这种运行时框架会非常合适。反过来说如果你完全不会写代码只能依赖界面配置那 Flowing 不适合你。我在项目文档里也把这句写在了 README 最前面省了双方的时间。另外Flowing 的定位是运行时而不是训练或评测工具所以它不像某些平台自带数据集和评测面板。你需要自己接监控。我目前的实践是每次请求结束后记录状态变更序列、每个步骤的耗时和 token 消耗写入日志系统这样出问题时可以回放完整交互过程。这个思路也是从智能体行为审计这个热门词里学到的——智能体应用跑起来容易但要让它稳定可靠一定要有可见的轨迹日志。6. 扩展思路如何让 Flowing 适配你现有的业务架构最后分享几个我实际验证过的集成方式。想让 Flowing 原生支持 SSE 流式输出无需额外封装。如果你的服务是内网 RPC 调用不需要 HTTP 接口可以直接在业务进程里通过agent.run_state(state)调用跟普通函数一样只是它可以异步。要接入多租户可以在状态里增加tenant_id字段然后在步骤里根据租户加载不同的工具配置。因为 Flowing 的状态对每个请求是独立的天然没有串号的问题。我目前的实践是每次请求结束后记录状态变更序列、每个步骤的耗时和 token 消耗写入日志系统这样出问题时可以回放完整交互过程。这个思路也是从智能体行为审计这个热门词里学到的——智能体应用跑起来容易但要让它稳定可靠一定要有可见的轨迹日志。如果希望把流式消息推给前端直接把agent.serve()里的 transport 换成 websocket 模式即可Flowing 内置了两种传输方式。我前端现在用的是fetch的流式读取后端这里直接用StreamingResponse实现只用了几分钟就接好了。最后说说我个人对智能体运行时这个方向的看法。现在智能体框架和平台层出不穷但真正能把复杂交互做好的并不多。复杂交互的本质是状态与流程的控制而不是堆多少个模型。运行时框架的价值是帮你把控制逻辑和工作流的通用模式沉淀下来让你自己的代码只关注业务本身。这正是我创建 Flowing 的初衷我也会持续在这个方向迭代。如果你也有类似的项目困扰欢迎顺着这个思路自己动手撸一个很多东西只有亲手跑过才知道坑在哪儿。 轻量级智能体运行时框架 Flowing：复杂交互场景下的设计与实践
1. 为什么我会自己折腾一个智能体运行时框架先交代一下背景。我在团队里主要负责把大模型能力落进实际业务差不多把市面上能见到的智能体平台和开源框架都摸过一圈之后发现一个尴尬的现状平台型的智能体搭建工具确实上手快但一旦涉及复杂的业务交互比如多轮状态保持、权限差异、动态工具路由、流式回包拼接平台那套可视化节点就开始力不从心而主流开源智能体框架功能全但重量也不小依赖一堆、抽象层层叠叠调试一个问题要翻好几层源码。我的需求其实很朴素要一个能写进现有服务里的智能体运行时不挑部署环境启动够快逻辑透明复杂交互我能自由编排。市面上没有完全合我胃口的于是就有了 Flowing 这个项目。它不是一个搭建平台而是一个运行时框架——你写代码来定义智能体的行为框架负责把大模型调用、工具执行、状态流转、多轮交互这些繁琐的底层逻辑收拢起来。这篇文章就把我设计 Flowing 时的核心思路、实际代码示例和踩过的坑都摊开讲如果你也有类似诉求可以参考着做出自己的选择。2. Flowing 的设计取舍轻量级不等于功能少而是去掉你不需要的部分轻量级这几个字被用滥了很多框架说自己轻量实际装完依赖一看体积比业务代码还大。我给 Flowing 定的标准很具体核心运行时没有任何必须的第三方大依赖只依赖标准库和一个轻量的 LLM 客户端接口定义。其他能力比如具体的模型厂商 SDK、向量存储、消息队列全部通过插件方式按需挂载。这样做的直接好处是你在一个只有几百 MB 内存的容器里也能跑起来而且每一次启动几乎是瞬时的没有一堆初始化检查。面向复杂交互这个定位决定了不能照着聊天机器人那套模式做。我拆解了常见的智能体交互场景抽象出三个核心原语状态State、事件Event、步骤Step。状态就是当前会话的上下文包括用户输入、中间变量、历史记录事件是外部触发的信号比如用户消息、定时任务、Webhook 回调步骤是智能体处理事件的最小单元可以是一个大模型调用、一段工具代码、一个条件判断也可以是另一个子智能体。框架的本质是一个按顺序执行步骤、并根据步骤结果更新状态的事件循环。这样做对比起图编排或链式调用的好处是你可以很容易地表达循环、嵌套和动态分支而这些正是复杂交互的刚需。很多人会问直接用 async 循环自己写不就行了吗确实可以但遇到重试、超时、上下文裁剪、并行执行几个子任务再汇总这些逻辑散落在业务代码里会非常啰嗦。Flowing 把这些横切关注点内聚到运行时里你的业务代码只需要描述做什么而怎么调度、怎么容错由框架兜底。3. 半小时跑通第一个 Flowing 智能体3.1 安装与项目结构Flowing 是一个 Python 包Python 版本要求 3.10 以上这主要是为了用上最新的类型语法和match语句。安装很简单pip install flowing-agent装完之后推荐按下面的目录结构组织你的项目这不算强制但我试过很多种布局后觉得这个最顺手my_agent/ ├── agent.py # 定义智能体的入口和步骤链 ├── steps/ # 业务相关的自定义步骤 │ ├── __init__.py │ └── fetch_order.py ├── tools/ # 智能体可调用的外部工具 │ └── query_db.py └── config.yaml # 模型、超时、日志等配置3.2 定义状态模型和第一个步骤第一步是定义状态。Flowing 内置了一个BaseState类你只需要声明字段运行时负责序列化和传递。为了支持复杂交互状态本身是可变对象每个步骤执行后都可以直接修改它。from flowing import BaseState class ChatState(BaseState): user_input: str user_id: str history: list[dict] intent: str | None None result: str 接着写一个步骤。步骤是实现Step接口的类关键是async def run(self, ctx)方法。ctx里能拿到当前状态、运行时上下文、以及一个markdown_text之类的辅助函数。我习惯把提示词模板也放在步骤类里这样做的好处是每个步骤自包含测试起来很方便。from flowing import Step, Context class UnderstandIntent(Step): async def run(self, ctx: Context[ChatState]): messages [ {role: system, content: 你是一个意图识别器只输出一个词查询、或执行。}, {role: user, content: ctx.state.user_input}, ] resp await ctx.llm.chat(messages) ctx.state.intent resp.text().strip() return self.OK3.3 组装智能体并启动服务Flowing 的智能体对象由一系列步骤组成运行时会按顺序执行。但不是死板的顺序步骤里可以通过返回状态码来提前中断或跳转这个机制后面会详细说。先看一个简单例子from flowing import Runtime, Step agent Runtime[ChatState]( stateChatState, steps[ UnderstandIntent(), SelectBranch(), CallToolOrReply(), ], # 这里传入你要接入的 LLM 客户端支持流式 ) if __name__ __main__: agent.serve(port8000)这样就得到了一个 HTTP 服务支持POST /chat接口请求体里{user_input: 帮我查一下昨天的订单, user_id: u_123}返回流式响应。整个过程没写任何 FastAPI 代码运行时直接把常用协议封好了。如果你不想起服务也可以在脚本里直接调用await agent.run_state(state)适合做单元测试或者批处理。4. 复杂交互的核心分支、循环、工具调用与子智能体这一章是 Flowing 真正区别于用 Python 手写代码的地方。我遇到过很多所谓智能体框架只能做线性的调一次 LLM - 执行一个工具的流程稍微复杂一点的业务就捉襟见肘。我设计 Flowing 时专门针对以下几个交互模式做了支持。4.1 带条件的动态路由业务里经常要按意图走不通的分支比如识别到查询就走检索步骤识别到下单就走校验步骤。Flowing 用步骤的next显式指定来支持路由同时允许条件表达式class SelectBranch(Step): async def run(self, ctx): if ctx.state.intent 查询: return self.GOTO(search_order) elif ctx.state.intent 执行: return self.GOTO(execute) return self.FAILGOTO的目标是步骤的名字运行时维护一个步骤注册表。你可以在Runtime初始化时命名步骤也可以让步骤类自己声明名字。这样流程就不是一条直线而是一张跳转表跳出循环和复杂状态机都方便。4.2 工具声明与自动调用智能体绕不开工具调用。Flowing 提供了一种简洁的工具注册方式不需要你手动解析大模型返回的 JSON 字符串from flowing import tool tool(description根据订单ID查询订单详情, params{order_id: string}) async def fetch_order(order_id: str) - dict: # 里面可以是任何查库、调外部 HTTP 的逻辑 result await query_db(order_id) return {order_id: order_id, status: result.status}在步骤里你可以显式调用这个工具也可以让大模型自主决定调用class ToolOrReply(Step): async def run(self, ctx): if ctx.state.intent 查询: tool_result await ctx.call_tool(fetch_order, order_id12345) ctx.state.result tool_result else: messages [{role: user, content: ctx.state.user_input}] ctx.state.result await ctx.llm.chat(messages) return self.OK自主调用模式需要 LLM 供应商支持 function calling 协议Flowing 的模型客户端接口里内置了工具描述传参你只需要在创建 LLM 客户端时传入工具列表。框架运行时会解析模型的函数调用请求自动执行工具并回填结果不用你手动拼接多轮回调消息。这一块当然需要与具体模型调接口但接口已经抽象得比较干净了像我实际用 OpenAI 兼容接口和通义千问的兼容接口都是改一行 base_url 就行。4.3 并行执行与结果合并复杂交互里经常要同时查多个数据源再汇总回答。Flowing 提供了一个FanOut步骤类型接收子步骤列表并行执行from flowing import FanOut class ParallelSearch(FanOut): steps [search_orders, search_products] async def aggregate(self, results): ctx.state.aggregated | .join(results) return self.OKaggregate在子步骤全部完成后被调用。并行执行用的是 asyncio 的gather所以你的子步骤里不能有阻塞式 IO如果有记得改成await asyncio.to_thread()。我踩过这个坑一开始在工具里直接用了requests结果并行直接变成了串行吞吐惨不忍睹后来统一改成httpx.AsyncClient才解决。4.4 子智能体把一个智能体看作一个步骤单个智能体管不了太巨大的任务Flowing 允许你在一个步骤里启动另一个智能体实例子智能体有自己的状态、步骤和模型配置。这样做在多角色协作场景里特别好用主智能体做任务拆解把具体环节交给搜索智能体、客服智能体执行再把结果汇总回主状态。实现上子智能体的 Runtime 实例就是一个普通的 Step因为它也需要输入输出。关键在于子智能体的状态隔离子智能体的状态是独立对象不能直接碰父智能体的内部字段只能通过输入参数传入和返回值传递结果。这个约束一开始我觉得麻烦但实际用下来避免了很多隐式耦合出了 bug 也好定位。5. 实际运行中的性能表现与避坑记录5.1 和常见框架的对比我拿我自己之前用过的典型平台和框架做了个不严谨的对比测试场景是 20 个并发用户、每个用户连续 10 轮多轮交互夹杂工具调用记录 P95 响应时长和框架初始化耗时从启动到可服务方案初始化耗时P95 响应依赖体积交互编排能力Flowing180ms2.1s核心约 1MB高原生支持跳转/循环/子智能体某全流程框架3.2s2.4s约 80MB中支持链式循环需要 hack某个平台依赖外部托管2.8s不适用中可视化编排但无法做复杂代码逻辑这个表格不是要说明 Flowing 比谁强而是想说对于已经熟悉代码、且业务逻辑有大量定制需求的团队一个轻量运行时带来的直接收益是启动快、好调试、依赖少。平台的定位是让不懂代码的人也能搭框架是让懂代码的人不受约束。两者的设计目标不同。5.2 踩过的三个坑状态序列化的问题。复杂状态里如果包含自定义对象保存会话快照时会挂。我后面统一规定状态字段只允许 JSON 可序列化类型自定义对象必须放步骤内部属性。这个规定虽然限制了一些自由度但换来了能方便地把状态存入 Redis 做持久化。流式输出的拼接与中断。大模型流式返回文本片段如果中途用户发来新消息需要打断当前响应我最初的处理是直接拿到当前 token 终止结果下一次恢复上下文时缺了半截话。后来我设计成取消当前步骤但把已经生成的 token 记录在状态里下一次响应先把这部分列出再开始新回复。稍微有点绕但用户体验好很多。工具调用的递归深度。某些场景下模型会连续调用十几个工具如果工具本身又触发新的子工具可能会无限递归。Flowing 里给每个运行实例加了一个max_tool_rounds参数默认 10超过直接报错防止把 token 烧光。5.3 什么时候用 Flowing什么时候不要用如果你只是做一个展示用的问答机器人直接调 LLM API 就够了不需要框架如果你想在一个大型平台里做复杂的业务流程且你团队里有能写 Python 的工程师那 Flowing 这种运行时框架会非常合适。反过来说如果你完全不会写代码只能依赖界面配置那 Flowing 不适合你。我在项目文档里也把这句写在了 README 最前面省了双方的时间。另外Flowing 的定位是运行时而不是训练或评测工具所以它不像某些平台自带数据集和评测面板。你需要自己接监控。我目前的实践是每次请求结束后记录状态变更序列、每个步骤的耗时和 token 消耗写入日志系统这样出问题时可以回放完整交互过程。这个思路也是从智能体行为审计这个热门词里学到的——智能体应用跑起来容易但要让它稳定可靠一定要有可见的轨迹日志。6. 扩展思路如何让 Flowing 适配你现有的业务架构最后分享几个我实际验证过的集成方式。想让 Flowing 原生支持 SSE 流式输出无需额外封装。如果你的服务是内网 RPC 调用不需要 HTTP 接口可以直接在业务进程里通过agent.run_state(state)调用跟普通函数一样只是它可以异步。要接入多租户可以在状态里增加tenant_id字段然后在步骤里根据租户加载不同的工具配置。因为 Flowing 的状态对每个请求是独立的天然没有串号的问题。我目前的实践是每次请求结束后记录状态变更序列、每个步骤的耗时和 token 消耗写入日志系统这样出问题时可以回放完整交互过程。这个思路也是从智能体行为审计这个热门词里学到的——智能体应用跑起来容易但要让它稳定可靠一定要有可见的轨迹日志。如果希望把流式消息推给前端直接把agent.serve()里的 transport 换成 websocket 模式即可Flowing 内置了两种传输方式。我前端现在用的是fetch的流式读取后端这里直接用StreamingResponse实现只用了几分钟就接好了。最后说说我个人对智能体运行时这个方向的看法。现在智能体框架和平台层出不穷但真正能把复杂交互做好的并不多。复杂交互的本质是状态与流程的控制而不是堆多少个模型。运行时框架的价值是帮你把控制逻辑和工作流的通用模式沉淀下来让你自己的代码只关注业务本身。这正是我创建 Flowing 的初衷我也会持续在这个方向迭代。如果你也有类似的项目困扰欢迎顺着这个思路自己动手撸一个很多东西只有亲手跑过才知道坑在哪儿。