人工智能大模型AI Agent工作流自动化RAG【免费下载链接】PocketFlowPocket Flow: 100-line LLM framework. Let Agents build Agents!项目地址https://gitcode.com/gh_mirrors/poc/PocketFlow点击查看免费下载本教程以 PocketFlow 官方 cookbook 中的pocketflow-gradio-hitl示例为蓝本讲解如何用 PocketFlow100 行级 LLM 框架编排带有人类反馈闭环的 Agent 工作流并通过 Gradio 提供可交互的 Web 界面。读完本文你将掌握如何设计决策式多节点 Flow、如何用共享上下文shared在节点间传递会话状态、如何把 LLM 的结构化 YAML 决策解析为流程跳转以及如何借助消息队列把 Agent 的思考过程实时流式渲染到聊天界面。示例概览一个会订酒店、查天气的生活助手cookbook/pocketflow-gradio-hitl是一个生活助理演示应用用户通过 Gradio 聊天界面与 AI 对话AI 能完成两类任务——查询指定城市与日期的天气、预订酒店含入住/退房日期当信息不足或问题超出范围时AI 会主动向用户追问任务完成后则向用户播报结果并询问是否还有其他需求。它的核心价值在于演示了Human-in-the-LoopHITL整个工作流不是一把梭式地调 LLM而是把决策与执行拆开——LLM 每轮只决定下一步动作需要用户补充信息如酒店名称、入住日期时就暂停流程向用户提问拿到反馈后再继续。这既避免了 LLM 一次生成错误参数也让每个动作都在人类监督下逐步推进。示例入口为 main.py流程编排在 flow.py节点实现集中在 nodes.py。快速上手安装、配置与启动1. 安装依赖先确认已克隆 PocketFlow 仓库并进入示例目录然后安装依赖见 requirements.txtpip install -r requirements.txt依赖项及版本下限如下与 README 中的 Requirements 一致依赖版本要求用途pocketflow 0.0.2工作流编排框架gradio 5.29.1Web 聊天界面5.x 起采用typemessages消息格式openai 1.78.1调用 OpenAI 模型2. 配置 OpenAI API Key应用使用 OpenAI 模型默认gpt-4o见 utils/call_llm.py做动作决策。设置环境变量export OPENAI_API_KEYyour-openai-api-key-here如果你用的是兼容 OpenAI 协议的代理或国内模型网关也可以直接修改utils/call_llm.py中的base_url与model两个常量示例默认值为https://api.openai.com/v1与gpt-4o。3. 启动应用python main.py启动后 Gradio 界面默认运行在http://localhost:7860。打开浏览器即可与 AI 对话示例会自动为每个会话生成一个 UUID 作为conversation_id用于隔离不同会话的状态见 main.py 中gr.State与clear_fn的实现。工作流架构五个节点与条件跳转整体流程图README 给出了完整的工作流拓扑Decide Action Node中枢决策节点。根据上一动作 动作执行结果 聊天历史 当前用户问题决定下一步动作。Check Weather Node查询指定城市与日期的天气。Book Hotel Node处理酒店预订请求含入住、退房日期。Follow Up Node向用户追问澄清信息或引导超出范围的请求。Result Notification Node向用户播报任务结果并提供进一步帮助。用 PocketFlow 的语法连接节点flow.py中通过 PocketFlow 的- action 与运算符完成带动作标签的条件连接与默认连接from pocketflow import Flow from nodes import ( DecideAction, CheckWeather, BookHotel, FollowUp, ResultNotification, ) def create_flow(): decide_action DecideAction() check_weather CheckWeather() book_hotel BookHotel() follow_up FollowUp() result_notification ResultNotification() decide_action - check-weather check_weather check_weather decide_action decide_action - book-hotel book_hotel book_hotel decide_action decide_action - follow-up follow_up decide_action - result-notification result_notification return Flow(startdecide_action)关键点node - action next_node表示条件边当节点post方法返回字符串action时流程跳转到next_node。node next_node表示默认边当post返回default时走这条路径。Flow(startdecide_action)以决策节点为起始。PocketFlow 的节点继承自BaseNode其__sub__返回携带动作标签的_ConditionalTransition对象__rshift__负责注册后继节点见 pocketflow/init.pyi 的类型声明。从源码结构看CheckWeather与BookHotel的post返回defaultnodes.py 与 nodes.py因此执行完天气查询或酒店预订后会沿默认边回到决策节点让 LLM 基于执行结果决定继续追问、再次调用工具还是播报结果——这正是循环型 Agent 工作流的典型形态。决策节点让 LLM 输出结构化 YAML 来驱动流程DecideAction是整条工作流的大脑它把下一步做什么完全交给 LLM 决策并通过强约束的 YAML 输出格式把决策结果结构化。prep组装上下文def prep(self, shared): conversation_id shared[conversation_id] session load_conversation(conversation_id) return session, shared[history], shared[query]prep从shared字典中读取会话 ID、聊天历史与当前用户问题并从会话缓存中加载last_action上一次动作与action_result上一次动作结果。exec构造提示词并解析决策exec中构造的提示词包含四部分角色指令你是能订酒店、查天气的生活助理需要基于上一动作、动作结果、聊天历史、当前问题决策下一步。聊天历史由format_chat_history格式化。上下文Last Action、Last Action Result、Current Date当前日期由datetime.now().date()动态注入。动作空间ACTION SPACE四个候选动作及其参数 schema动作适用场景关键参数check-weather用户询问天气city必填、date选填缺省用当天book-hotel用户要订酒店hotel、checkin_date、checkout_date均必填follow-up问题超出范围或当前信息不足以调用工具question追问内容result-notification酒店预订或天气查询已完成result结果播报提示词强制 LLM 输出固定格式的 YAMLthinking: | 你的逐步推理过程 action: check-weather OR book-hotel OR follow-up OR result-notification reason: 选择该动作的原因 question: 当动作是 follow-up 时 city: 当动作是 check-weather 时 hotel: 当动作是 book-hotel 时 checkin_date: 当动作是 book-hotel 时 checkout_date: 当动作是 book-hotel 时 result: 当动作是 result-notification 时并在提示词末尾强调三条格式纪律多行字段用 4 空格缩进、多行文本用|块引用、单行字段不用|。随后调用call_llm获取响应通过response.split(yaml)[1].split()[0].strip()剥离代码围栏再用yaml.safe_load解析为字典。post写回会话并决定下一跳post是 HITL 闭环的关键def post(self, shared, prep_res, exec_res): conversation_id shared[conversation_id] session load_conversation(conversation_id) session[last_action] exec_res[action] flow_log shared[flow_queue] for line in exec_res[thinking].split(\n): line line.replace(-, ).strip() if line: flow_log.put(f {line}) if exec_res[action] check-weather: session[check_weather_params] { city: exec_res[city], date: exec_res.get(date, None), } flow_log.put(f➡️ Agent decided to check weather for: {exec_res[city]}) elif exec_res[action] book-hotel: session[book_hotel_params] { hotel: exec_res[hotel], checkin_date: exec_res[checkin_date], checkout_date: exec_res[checkout_date], } flow_log.put(f➡️ Agent decided to book hotel: {exec_res[hotel]}) elif exec_res[action] follow-up: session[follow_up_params] {question: exec_res[question]} flow_log.put(f➡️ Agent decided to follow up: {exec_res[question]}) elif exec_res[action] result-notification: session[result_notification_params] {result: exec_res[result]} flow_log.put(f➡️ Agent decided to notify the result: {exec_res[result]}) save_conversation(conversation_id, session) return exec_res[action]这里发生了三件事持久化动作参数把决策中的city/date、hotel/checkin_date/checkout_date、question、result分别写入会话 session供后续执行节点取用。推送流程日志把 LLM 的thinking逐行、以及动作摘要➡️ ...推入flow_queue供前端实时展示思考过程。返回动作名作为跳转条件return exec_res[action]的值会与flow.py中注册的动作标签匹配从而驱动条件边跳转——check-weather去CheckWeatherbook-hotel去BookHotelfollow-up去FollowUpresult-notification去ResultNotification。执行节点天气查询与酒店预订CheckWeather / BookHotel两个执行节点的结构完全对称都遵循 PocketFlow 的prep → exec → post三段式prep从 session 中取出上一步决策写入的参数城市/日期或酒店/入住/退房日期。exec调用 mock APIutils/call_mock_api.py返回模拟结果避免真实调用外部服务。post把结果推入flow_queue⬅️ ...写入session[action_result]然后返回default让流程回到决策节点。Mock API 内置了贴近真实业务的校验规则是很好的参数边界参考天气查询日期与当前日期相差超过 7 天返回失败abs(date_diff) 7。酒店预订入住日期必须晚于当前日期入住日期不得晚于等于退房日期入住天数不得超过 7 天。模拟返回值天气为sunny/cloudy/rainy/snowy随机选择、温度10~30°C随机酒店预订成功返回包含日期范围的确认信息。这些规则保证了 LLM 即使脑补了参数也会在执行层被校验拦截并反馈回决策节点触发 follow-up 追问这正是 HITL 的意义所在。FollowUp / ResultNotification通过消息队列与前端通信这两个节点不再直接操作 LLM而是通过shared[queue]一个 Pythonqueue.Queue把要展示给用户的最终文本发送给 Gradio 前端class FollowUp(Node): def prep(self, shared): flow_log shared[flow_queue] flow_log.put(None) # 标记思考日志结束 conversation_id shared[conversation_id] session load_conversation(conversation_id) question session[follow_up_params][question] return question, shared[queue] def exec(self, prep_res): question, queue prep_res queue.put(question) queue.put(None) # 标记聊天消息结束 return question def post(self, shared, prep_res, exec_res): conversation_id shared[conversation_id] session load_conversation(conversation_id) session[action_result] exec_res return done注意post返回done而flow.py中没有注册done边因此流程在FollowUp/ResultNotification节点处自然终止——用户的每条消息触发一轮完整的决策→执行→可能追问/播报循环而追问的答案由用户在下一条消息中给出。ResultNotification在播报后还会清空action_result与last_actionnodes.py为下一轮对话重置上下文。会话管理shared 与内存级会话缓存PocketFlow 的节点之间通过shared字典共享数据。在本示例中main.py 为每轮对话构造了shared { conversation_id: str(uuid), query: message, history: history, queue: chat_queue, # 用户可见的聊天消息 flow_queue: flow_queue, # 思考过程/流程日志 }而跨节点、跨轮次的状态last_action、action_result、各动作参数则由 utils/conversation.py 提供的内存缓存管理conversation_cache {} def load_conversation(conversation_id: str): return conversation_cache.get(conversation_id, {}) def save_conversation(conversation_id: str, session: dict): conversation_cache[conversation_id] session以conversation_idUUID为键隔离不同会话避免多人同时使用时状态串扰。需要说明的是这是单进程内存缓存重启进程后状态即丢失如需持久化可以替换为文件存储或 Redis 等方案。此外utils/format_chat_history.py 在把历史喂给 LLM 前会过滤掉以- 、- ➡️、- ⬅️开头的思考/流程日志行防止 Agent 把自己的思维链当成对话内容重复输入。前端集成Gradio 多线程流式渲染main.py 展示了如何把 PocketFlow 同步流程嵌入 Web 应用全局线程池ThreadPoolExecutor(max_workers5)让多个会话的 Flow 并发运行避免阻塞 Gradio 事件循环。异步提交chatflow_thread_pool.submit(chat_flow.run, shared)把 Flow 丢到后台线程执行。双队列流式输出主线程先消费flow_queue累积思考日志逐条yield更新ChatMessage其metadata含titleFlow Log、status与duration字段待流程结束队列收到None哨兵后再消费chat_queue把追问或结果作为普通消息yield出去。界面搭建gr.Blocks(fill_heightTrue, themeocean)中放置gr.Chatbot(typemessages)与gr.ChatInterface通过gr.State保存 UUID点击清除按钮时clear_fn生成新 UUID 开启全新会话。这样用户能在界面上实时看到 Agent 的思考步骤与节点执行顺序而不仅仅是最终答案。运行效果与流程可视化应用提供实时流程可视化节点激活顺序按时间先后展示用户可以看到哪些决策路径被触发从而理解 AI 的决策过程。下图是用户发送hello时的完整界面上方是带耗时统计的Flow Log逐条展示 Agent 的分析与决策用户用 hello 问候无天气或酒店预订的上下文需友好跟进以明确需求下方是 AI 的友好追问预订酒店的示例输出——用户提出明天订酒店AI 追问酒店与入住时长发现最长只能订 7 天后与用户协商最终确认纽约希尔顿 2025-05-31 至 2025-06-07 的预订对话中途改变意图的示例——用户从订酒店临时转向查纽约明天天气AI 先给出天气结果再确认继续预订希尔顿 3 天展示了 HITL 工作流对用户意图变化的灵活响应文件结构与扩展建议示例目录结构如下文件作用main.py入口Gradio 界面搭建、线程池、双队列流式输出flow.py定义 PocketFlow 图与节点连接关系nodes.py五个节点的实现utils/call_llm.pyOpenAI 调用封装模型、base_url 可改utils/call_mock_api.py天气/酒店的模拟 API 与业务校验utils/conversation.py内存级会话缓存utils/format_chat_history.py聊天历史格式化过滤思考日志requirements.txt依赖清单想扩展这个示例时推荐按以下顺序动手新增工具节点在ACTION SPACE中追加新动作描述与参数 schema在flow.py中注册条件边在nodes.py中实现新节点并在post中保存参数。替换 mock API把call_book_hotel_api/call_check_weather_api换成真实后端 HTTP 调用业务校验可保留在服务端。持久化会话将conversation_cache换成数据库或 Redis即可支持多实例部署。升级异步如需更高并发可参考仓库内 docs/core_abstraction/async.md 与 tests/test_async_flow.py将节点改为AsyncNode。小结pocketflow-gradio-hitl用约 300 行代码完整演示了决策循环 人工介入 可视化的 HITL 应用范式LLM 只做结构化决策工具执行与参数校验交给普通节点用户追问通过消息队列回到前端整个流程由 PocketFlow 的shared上下文与动作标签无缝串联。这套模式稍加改造即可复用于预订、下单、审批等任何AI 出主意、人来拍板的真实业务场景。赞分享人工智能大模型AI Agent工作流自动化RAG【免费下载链接】PocketFlowPocket Flow: 100-line LLM framework. Let Agents build Agents!项目地址https://gitcode.com/gh_mirrors/poc/PocketFlow点击查看免费下载相关推荐英雄联盟智能助手用League Akari彻底改变你的游戏体验英雄联盟智能助手用League Akari彻底改变你的游戏体验 你是否曾经因为错过对局接受而懊恼是否在英雄选择阶段手忙脚乱是否想在对局开始前就了解队友和对人工智能大模型AI Agent工作流自动化RAGApache Airflow HITLHuman-in-the-loop实战指南用人工审批与决策操作符为工作流注入人类判断Apache Airflow HITLHuman in the loop实战指南用人工审批与决策操作符为工作流注入人类判断 Human in the Lo后端任务调度工作流自动化数据编排批处理数据工程流程编排Agent Governance Toolkit 审批工作流实战为高风险 Agent 动作构建 Human-in-the-Loop 审批门Agent Governance Toolkit 审批工作流实战为高风险 Agent 动作构建 Human in the Loop 审批门 导读 本文基于人工智能AI AgentAI 安全治理策略引擎Agent 沙箱认证鉴权上一篇【亲测免费】 tsParticles 使用教程下一篇【亲测免费】 Vue-Vben-Admin现代化的Vue.js后台管理系统搭建教程创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
