从 ReAct 到 Workflow：基于云端 API 构建事件驱动的智能体

1. 什么是WorkFlow之前咱们的用法是一种QueryEngine的用法就是将大模型当成一个查询的工具在使用而workflow是LlmaIndex的新一代编排引擎。1.1 核心逻辑LlamaIndex的workflow本质上是一个事件驱动Event-driven的状态机如果各位友人们经常做开发的话一定一下就理解。 咱们可以将其流程想象成一场接力赛:选手Steps 我们会在代码里用step装饰函数这个函数就是step负责具体的逻辑。接力棒Events我们会用来判断状态的自定义类继承自Event用于在Step间传递数据。终点线StopEvent 和普通的Event一样只是作为最后一棒某个Step返回StopEvent就意味着接力赛结束了咱们将结果返回给用户。1.2 核心组件Event是Workflow的灵魂咱们可以定义各种Event,其决定了数据流向何处。from llama_index.core.workflow import Event class NewsExtractedEvent(Event): content: str # 这里可以随便定义内容 xxx: int # 多少条都可以step是逻辑处理单元。一个Step可以通过返回值来“发射”事件。生产者 返回一个Event实例。消费者 在接收的下一步Step函数的函数声明参数中加上对应的Event实例。StartEvent StopEvent是流程的起跑线和终点线。一般我们只用定义StopEvent当调用workflow.run()自动会触发StartEvent任何Step只要返回了StopEvent整个workflow就会停止并输出结果。结合起来就是这样一个极简的Workflow模板from llama_index.core.workflow import Workflow, step, StartEvent, StopEvent, Event class ProcessedDataEvent(Event): data: str class MyWorkflow(Workflow): step async def ingest(self, ev: StartEvent) - ProcessedDataEvent: # 1. 接收初始输入 input_data ev.get(input) return ProcessedDataEvent(datainput_data.upper()) step async def finish(self, ev: ProcessedDataEvent) - StopEvent: # 2. 接收上一步处理后的数据 return StopEvent(resultf处理完成: {ev.data}) # 运行 wf MyWorkflow() await wf.run(inputhello world) # - 处理完成: HELLO WORLD通常来讲咱们会把step定义给一个调用llm处理的函数. 我们待会儿就用Workflow重写咱们上一篇中的Agent。2. API获取咱们这次给大家换换口味不用本地加载模型而是用API的方式去调用大模型所以本篇博客的代码可以在没有显卡的电脑上运行可以脱离Kaggle的显卡环境。PS: 虽然调用API更方便但是本地加载或者本地部署更安全阿尔周末也尝试过在kaggle上直接去用vllm/ollama部署然后调用但是遇到了太多没法解决的问题本次就直接用现成的API了。如果能解决到时候再出一篇教程吧如果还记得这个事儿的话哈哈哈接下来介绍两个可以用免费模型的APIGroq和硅基流动。2.1 GroqGroq并不是模型开发者只是大模型的搬运工是一个推理加速平台其核心是自研的LPU(Language Processing Unit),所以它的推理速度那是相当惊人这也是阿尔推荐它的很重要的一个理由。Groq模型矩阵模型系列模型 ID (API 调用名)厂商上下文窗口特点与优势Llama 3.3llama-3.3-70b-versatileMeta128k最强逻辑性能对标 GPT-4oLlama 3.1llama-3.1-8b-instantMeta128k极速响应适合简单逻辑与翻译Llama 3.2llama-3.2-11b-vision-previewMeta128k多模态支持可识别并理解图片Gemma 2gemma2-9b-itGoogle8k指令遵循强结构化输出更稳定Mixtralmixtral-8x7b-32768Mistral32k经典的混合专家模型长文本表现好Whisperwhisper-large-v3OpenAI-极速语音转文字 (STT)Groq的限制Groq的免费层级给的非常慷慨堪称大善人但是为了防止滥用还是设置了很多限制。通常对每个模型独立计算通常从RPM(Requets Per Minute)每分钟请求数RPDRequests Per Day每天请求数TPMTokens Per Minute每分钟消耗Token数三个维度限制当然对于咱们来说是完全够用的注册Groq Api Key进入Groq官网点击右上角的Start Building用谷歌账号登录即可点击最上面菜单栏的API KEYS点击Create API Key可以设置一个过期日期然后保存下来待会儿咱们配置到kaggle的Secrets中即可生成API Key2.2 硅基流动硅基流动是咱们要介绍的二号大模型的搬运工国内开发者使用比较多可以在上面找到很多便宜大碗的模型也有不少免费的模型可以用咱们今天就要用它的免费模型。硅基流动模型矩阵模型系列最新代表型号 (API 调用名)核心优势上下文窗口状态DeepSeekdeepseek-ai/DeepSeek-V3.2性能对标顶级闭源模型极强逻辑128k旗舰收费DeepSeekdeepseek-ai/DeepSeek-R1推理增强模型擅长复杂数学与逻辑64k部分免费QwenQwen/Qwen3-30B-A3B阿里最新一代模型中文语境指令遵循极强128k免费/收费Qwen-VLQwen/Qwen3-VL-72B顶级视觉理解能力支持视频分析32k旗舰收费GLMZ.ai/GLM-5智谱旗舰专注于长周期 Agent 任务205k旗舰收费Kimimoonshotai/Kimi-K2.5原生多模态智能体模型长文本处理极佳262k旗舰收费Stepstepfun-ai/Step-3.5-Flash极高性价比的轻量模型推理速度极快262k低成本/免费MiniMaxMiniMaxAI/MiniMax-M2.5擅长角色扮演与情感交互197k旗舰收费Llama 3meta-llama/Llama-3.3-70B-Instruct国际最强开源模型适配各类 Agent 框架128k旗舰收费硅基流动的限制硅基流动有着比Groq更多的可用模型有不少永久免费的模型, 甚至还有推广阿尔这里就不贴推广码了大家自行注册即可和新用户赠金但是其免费版的并发数很低具体的模型也有RPM和TPM限制。注册硅基流动API Key进入硅基流动官网用手机号或者微信号注册点击左侧侧边栏的API密钥, 点击新建API密钥我们待会儿使用Qwen/Qwen3-8B这个免费模型3. 代码实战3.1 环境准备这次和此前不同诶咱们今天用API不用自己加载模型不用开GPU加速Accelerator选择None即可。然后老样子先装依赖库# 1. 确保安装最新版 uv !pip install -U uv # 2. 使用修正后的包名进行安装 # llama-index 包含核心组件 # llama-index-llms-groq 是 Groq 的适配器 # llama-index-llms-llms-openai-like 是 所有适配openai接口的适配器 !uv pip install llama-index llama-index-llms-groq llama-index-llms-openai-like feedparser requests transformers注意这里咱们硅基流动其实也是有专门的定制库的llama-index-llms-siliconflow只是这里咱们为了通用性用了llama-index-llms-openai-like。实际上是因为最开始试了Groq效果不好觉得换中文模型会好一点然鹅效果也不好这次主要还是介绍这个WorkFlow架构吧3.2 配置密钥和上篇博客一样咱们在Add-ons-Secrets中添加上Groq和硅基流动的API Key命名为GROQ_API_KEY和SILICON-API_KEY当然名字无所谓和代码匹配就好3.3 使用API来示例llmTools和之前一样咱们就不赘述了。定义Groq的模型import os from llama_index.llms.groq import Groq from llama_index.core import Settings os.environ[GROQ_API_KEY] user_secrets.get_secret(GROQ_API_KEY) # 推荐使用 llama-3.3-70b它的逻辑推理能力足以应对多智能体协作 Settings.llm Groq(modelllama-3.3-70b-versatile)定义硅基流动的Qwen模型from llama_index.llms.openai_like import OpenAILike Settings.llm OpenAILike( modelQwen/Qwen3-8B, api_keyuser_secrets.get_secret(SILICON-API_KEY), api_basehttps://api.siliconflow.cn/v1, is_chat_modelTrue )可以看出来定制的接口只用定义对应的API KEY在环境变量里通用的接口需要补充清楚模型名api-keyapi的url等等。3.4 整体流程我们打算定义两个AgentChecker: 负责去调用获取最新博客以及解析Notifier: 负责向我们发送通知当有新的博客的时候Checker向Notifer发送事件告知notifier新博客的标题和链接。然后结束。整体流程如下3.5 定义Eventfrom llama_index.core.workflow import Workflow, step, StartEvent, StopEvent, Context, Event class NewBlogPostEvent(Event): title: str link: str由于Checker流转向Notifer的时候只需要给它标题和链接即可咱们的NewBlogPostEvent只用包含title和link字段。3.6 定义两个合作的Agent我们先定义Checker交代好它的角色ROLE任务TASK和输出要求OUTPUT_FORMAT将checker需要调用的工具给它from llama_index.core.agent.workflow import ReActAgent CHECKER_SYSTEM_PROMPT ### ROLE 咱们是一个极其精准的数据提取专家。 ### TASK 1. 调用工具获取最新博客文章。 2. 提取其标题和链接。 3. **必须**以 JSON 格式输出。 ### OUTPUT_FORMAT {title: 文章标题, link: 文章链接} checker ReActAgent( namechecker, system_promptCHECKER_SYSTEM_PROMPT, tools[tool_get_blog], llmSettings.llm )然后咱们再来定义NotifierNOTIFIER_SYSTEM_PROMPT ### ROLE 咱们是一个通知官。 ### TASK 请根据提供的信息发送通知。 直接调用工具发送不要解释咱们的行为。发送成功后回复“已发出”即可。 notifier ReActAgent( namenotifier, system_promptNOTIFIER_SYSTEM_PROMPT, tools[tool_send_all], llmSettings.llm )3.7 定义Workflow从Workflow类派生出我们的workflow类定义step从StartEvent开始到StopEvent结束这里因为模型的指令跟随能力比较差格式化输出不了咱们用了一个正则来兜底。import os import re import json global_last_title class DemoBlogMonitor(Workflow): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.checker checker self.notifier notifier step async def check_step(self, ev: StartEvent) - NewBlogPostEvent | StopEvent: global global_last_title rss_url ev.get(rss_url) print(f--- [Checker] 目标: {rss_url} ---) print(f--- [Checker] 基准: {global_last_title} ---) response await self.checker.run(user_msgf请提取此 RSS 的最新数据{rss_url}) res_text str(response) print(fDEBUG [Agent Output]: {res_text}) try: json_match re.search(r(\{.*\}), res_text, re.DOTALL) if not json_match: raise ValueError(未在回复中找到 JSON 块) data json.loads(json_match.group(1)) current_title data.get(title, ).strip() current_link data.get(link, ).strip() if not current_title or not current_link: raise ValueError(JSON 数据缺失字段) # 在 Python 侧进行确定性比对 if current_title ! global_last_title: print(f发现新动态: {current_title}) # 注意这里我们只在成功触发通知时更新状态 return NewBlogPostEvent(titlecurrent_title, linkcurrent_link) else: return StopEvent(resultf判定内容未更新) except Exception as e: # 如果 JSON 解析失败尝试最后的正则兜底 print(fWARN: JSON 解析失败 ({e}), 尝试正则兜底...) link_match re.search(r(https?://\S), res_text) if link_match: link link_match.group(1).strip( ”\。) # 假设引号中间的内容是标题 title_match re.search(r[是“\](.*?)[”\], res_text) title title_match.group(1) if title_match else 未识别标题 if title ! global_last_title: return NewBlogPostEvent(titletitle, linklink) return StopEvent(resultf解析彻底失败。原文: {res_text}) step async def notify_step(self, ev: NewBlogPostEvent) - StopEvent: global global_last_title print(f--- [Notifier] 正在发送: {ev.title} ---) await self.notifier.run(user_msgf发送通知标题《{ev.title}》链接 {ev.link}) # 只有通知成功发出后才真正更新全局基准 global_last_title ev.title return StopEvent(resultf完成。新基准已设为: {ev.title})3.8 运行async def main(): monitor DemoBlogMonitor(timeout120) print(\n Run #1) await monitor.run(rss_urlhttps://blog.algieba12.cn/atom.xml) print(\n Run #2) await monitor.run(rss_urlhttps://blog.algieba12.cn/atom.xml) await main()这里咱们运行两遍确认全局变量会同步更新到新状态咱们实际使用的时候可以将其持久化存储下来而不是用变量global_last_title将状态保存在内存中4. 代码本篇博客代码示例可在Kaggle笔记本5. 常见问题 (QA)Q1: 为什么不让 Agent 直接判断是否有更新而是要在 Python 代码里写if current_title ! global_last_title:A:为了系统的绝对稳定性(划重点)大模型在 ReAct 模式下有时会产生“幻觉”或者表现得过于“客气”。现象最开始的设计是让模型自己比对已知标题和新标题结果模型看到传进去的旧标题占位符误以为这是一道“填空题”不仅没有进行比对还很有礼貌地让我把真实标题填进去。结论最佳实践是逻辑解耦。让 Agent 专注于它最擅长的事情从杂乱网页中精准提取非结构化数据而将确定性的逻辑判断字符串比对交还给传统的 Python 代码来做。Q2: 为什么一定要强制模型输出 JSON 格式之前用|这种自定义分隔符不行吗A:因为自然语言的边界太模糊且极易与真实数据冲突。这是本次开发多智能体协作时遇到的最大痛点表达欲失控高级模型比如 Llama-3.3 或 Qwen即使在 Prompt 里千叮咛万嘱咐只输出数据也极大概率会在结果前后加上“好的最新文章是…”之类的废话导致split(|)直接报错。特殊符号冲突碰巧咱们博客的标题里本身就带有|符号。正则匹配遇到这种情况会产生严重的边界歧义。JSON 的优势JSON 拥有明确的边界标识花括号和双引号。再配合 Python 正则re.search(r(\{.*\}))专门抓取大括号之间的内容就能彻底免疫模型的“废话”。Q3: 运行 Workflow 时碰到了WorkflowValidationError: The following events are produced but never consumed报错是怎么回事A:这是 LlamaIndex 工作流引擎的一种“管道安全检查”机制。它的意思是咱们的check_step辛辛苦苦生产了一个NewBlogPostEvent但是工作流里没有任何一个 Step 声明要接收并处理这个事件导致事件“无家可归”。排查方法检查咱们的消费者函数在这个例子中是notify_step。解决方案确保其参数严格加上了对应的类型注解写成async def notify_step(self, ev: NewBlogPostEvent)。这样引擎就能自动把上一步产生的数据路由到这里完成闭环。Q4: 调用 Groq 或硅基流动 API 时频繁遇到429 Too Many Requests怎么办A:这是因为多智能体循环请求过快触发了并发限制。在开发 ReAct Agent 时模型需要反复执行“思考-行动-观察”的循环短时间内会产生高密度的 API 请求。冷却机制如果是演示项目可以在每次检测循环之间或者 Step 之间加一个简单的await asyncio.sleep(2)进行冷却。精简上下文尽量精简 Prompt 和工具返回的数据量避免单次请求打满 TPMTokens Per Minute限制。如果是生产环境建议使用硅基流动的付费版以获取更高的并发额度。Q5: 在生产环境中我还应该使用global global_last_title这种全局变量来管理状态吗