这篇我们来说说共享状态、共享数据以供后续步骤使用如用户信息、累计值一、状态管理:让步骤之间共享数据1.1 为什么要状态管理?有些数据是整个 Workflow 共享的,比如:用户信息(从第一步输入,后面每步都要用)累积的统计数据(每步加一点)缓存的中间结果这种全局共享数据就放在Context里。1.2 Context 基础 APIfrom llama_index.core.workflow import Context # 写 await ctx.set(user_name, 张三) await ctx.set(counter, 0) # 读 user_name await ctx.get(user_name) # 张三 counter await ctx.get(counter, default0) # 带默认值 # 自增 counter await ctx.get(counter, default0) await ctx.set(counter, counter 1)1.3 完整示例:多步累积计算class AccumulateWorkflow(Workflow): step async def init(self, ctx: Context, ev: StartEvent) - NumberEvent: 初始化 state,接收输入 await ctx.store.set(total, 0) await ctx.store.set(count, 0) return NumberEvent(numbers ev.numbers) step async def accumulate(self, ctx: Context, ev: NumberEvent) - StopEvent: 每收到一个数字就累加,所有数字走完就停 total await ctx.store.get(total) count await ctx.store.get(count) for n in ev.numbers: total n count 1 print(f当前累加: {total},已处理 {count} 个数) await ctx.store.set(total, total) await ctx.store.set(count, count) return StopEvent(result f总和: {total},数量: {count})运行结果:1.4 关键:step 怎么拿到 ctx?async def accumulate(self, ctx: Context, ev: NumberEvent) - StopEvent: # ^^^ ^^^ # 第一个参数 第二个参数记住这个顺序:ctx永远是第一个参数(放在self后面),ev是事件参数(紧跟其后)。框架通过类型注解Context来识别这个参数,自动注入不用手动Context(workflow)1.5 状态序列化:跨进程持久化Context 状态可以序列化到磁盘,实现程序重启后状态还在:# 保存 state await ctx.to_dict() import json with open(workflow_state.json, w) as f: json.dump(state, f) # 恢复(在新的 Workflow 实例中) with open(workflow_state.json, r) as f: state json.load(f) new_ctx Context(WorkflowClass) await new_ctx.from_dict(state)这个特性在做长任务(比如几个小时的数据处理)时特别有用——中途崩了可以从断点继续。二、状态 vs 事件:啥时候用哪个?这是新手最容易混的地方:简单判断:数据只在一个 step 里用 →用局部变量数据在相邻两步之间传 →用事件数据多个 step 共享 →用状态三、组合实战:分支 循环 状态来个真实点的例子:一个猜数字Workflow,数字不对就一直猜,猜对就停。class GuessNumberWorkflow(Workflow): step async def init(self, ctx: Context, ev: StartEvent) - GuessEvent: 初初始化:把目标数字存到 state target ev.target or random.randint(1, 100) # 设定初始搜索区间 low 1 high 100 # 如果用户提供了第一次猜测直接使用否则取中点 if ev.first_guess is not None: first_guess ev.first_guess else: first_guess (low high) // 2 # 存储到上下文 await ctx.store.set(target, target) await ctx.store.set(low, low) await ctx.store.set(high, high) await ctx.store.set(attempts, 0) return GuessEvent(guess first_guess, feedback 开始猜数) step async def accumulate(self, ctx: Context, ev: GuessEvent) - StopEvent|GuessEvent: 每收到一个数字就累加,所有数字走完就停 target await ctx.store.get(target) attempts await ctx.store.get(attempts, default 0) low await ctx.store.get(low) high await ctx.store.get(high) attempts 1 await ctx.store.set(attempts, attempts) if ev.guess target: return StopEvent(result f 猜对了!用了 {attempts} 次) elif ev.guess target: feedback 太小了 low max(low, ev.guess 1) # 目标在 guess1 到 high 之间 else: feedback 太大了 high min(high, ev.guess - 1) # 目标在 low 到 guess-1 之间 if attempts 10: return StopEvent(result f 超时失败,目标数字是 {target}) # 继续猜:返回 GuessEvent 又回到自己 next_guess (low high) // 2 # 二分法 print(f第 {attempts} 次猜 {ev.guess}{feedback}区间 [{low}, {high}]下次猜 {next_guess}) # 更新存储的区间 await ctx.store.set(low, low) await ctx.store.set(high, high) return GuessEvent(guess next_guess, feedback feedback)运行会看到:四、新手最常踩的坑坑 1:状态忘了初始化# ❌ 错误:没初始化就 get step async def step1(self, ctx: Context, ev: StartEvent): total await ctx.get(total) # KeyError! ... # ✅ 正确:get 时带默认值,或者先 set total await ctx.get(total, default0)坑 2:Context 的坑——并发写会覆盖# ❌ 错误:多个 step 同时写同一个 key step async def step_a(self, ctx: Context, ev: Ev) - Ev: await ctx.set(counter, 10) ... step async def step_b(self, ctx: Context, ev: Ev) - Ev: await ctx.set(counter, 20) # 可能覆盖 step_a 的写入 ...坑 3Context赋值问题# 此方式已过时 await ctx.set(counter, 10) # 应该使用 await ctx.store.set(counter, 10)DeepSeek和豆包都没找到问题所在minmax直接找到问题点五、本篇小结这一篇我们把 Workflow 从动态业务引擎升级成了可共享数据的动态业务引擎:循环:让 step 返回上一步的事件类型,形成事件回路状态管理:ctx.get()/ctx.set()共享全局数据状态序列化:ctx.to_dict()/ctx.from_dict()跨进程持久化俩大坑:状态要初始化、并发要避免覆盖核心心法:事件是接力棒,状态是记录本——事件负责动,状态负责存。下一篇我们讲流式输出与并发执行——让 Workflow 边跑边吐结果、多个分支同时跑,大幅提升性能和用户体验。