无模型智能体：用LangGraph构建确定性状态工作流

1. 项目概述为什么一个“没有大模型”的工作流反而更像真正的智能体你有没有试过调试一个用 LLM 驱动的所谓“智能体”输入一发输出一团中间发生了什么没人知道。你改了提示词它偶尔好一点偶尔更糟你加了个工具调用结果它在不该调用的时候调用了三次你设了个重试逻辑它却在验证失败后直接返回了“抱歉我无法处理”而不是去补全缺失字段——整个过程像在和一个情绪不稳定的实习生打交道你得不断猜它脑子里在想什么再反过来倒推怎么写 prompt 才能把它“哄”回正轨。这篇文章讲的恰恰是反其道而行之先彻底拿掉语言模型只用 LangGraph 搭一个纯确定性的、连一行生成式代码都没有的工作流却让它展现出比很多“LLMAgent”系统更清晰、更可控、更可调试的智能体行为。关键词不是“大模型”“推理”“生成”而是状态State、路由Routing、重试Retry、收敛Converge和可观测性Observability。它解决的不是一个炫技型问题而是一个真实业务中天天发生的场景客户订单出了问题你得快速判断是货损、发货延迟还是发错货再决定走退款还是换货流程——但客户提交的信息往往缺这少那格式也不对系统不能崩也不能瞎猜必须一步步把信息收齐、验准、分路、办结。我做智能体系统落地三年从金融风控到电商售后踩过最深的坑不是模型不准而是结构没立住。当团队急着把 GPT-4 接进 pipeline 时我坚持先用 Python 字典 if/else 写完整个状态流转逻辑再把每个判断点替换成模型调用。这个习惯救了我们两次上线危机一次是某天凌晨订单量突增十倍模型 API 延迟飙升但我们的状态机依然稳稳地在重试队列里排队、校验、分发另一次是客户投诉“系统自己改了我的退款金额”我们打开状态快照回放三分钟定位到是某个节点误把字符串100.00当整数加了1而不是模型胡说八道。所以这篇看似“不实用”的纯结构实践其实是所有 agentic 系统的底盘工程——它不告诉你模型怎么写 prompt但它决定了你的系统在模型宕机、输入脏乱、流程分支爆炸时还能不能呼吸、还能不能思考、还能不能被你亲手拉回来。适合两类人一类是刚接触 LangGraph、被“agent”二字吓住的新手另一类是已经用上 LLM 却总在 debug 时抓狂的工程师。接下来我们就从零开始把这套“无模型智能体”的筋骨一节节拆开给你看。2. 核心设计思路为什么“去掉LLM”不是妥协而是战略聚焦2.1 从“生成幻觉”到“执行确定性”的范式切换绝大多数关于智能体的教程开篇就是“让我们加载一个 Llama3 模型然后用 LangChain 的 AgentExecutor 封装它……” 这种路径天然带着一个危险预设智能 生成能力。于是整个系统的设计重心就滑向了如何让模型“说得像人”“答得全面”“逻辑自洽”。但现实中的业务系统真正要命的从来不是“说得好不好”而是“做得对不对”。举个具体例子客户提交了一个order_id: ORD-123和issue_type: damaged但漏填了resolution_preference。一个纯 LLM 驱动的 agent 可能会这样响应“检测到您反馈商品损坏为保障您的权益我们将为您安排换货服务。请稍候系统正在处理……”——它甚至没意识到resolution_preference是必填项直接跳过了验证环节用“默认换货”代替了“主动询问”。这不是聪明这是失控。而我们即将构建的结构化工作流第一步就强制要求所有字段必须显式声明、显式校验、显式缺失标记。当resolution_preference为空时系统不会生成任何响应而是将is_valid设为Falseerrors数组里塞进resolution_preference is required然后路由回input节点——这个动作不是靠模型“理解”出来的而是由route_after_validation函数里一行if not state.get(is_valid): return retry_input硬编码决定的。它的确定性来自你写的代码而不是模型的概率分布。这种设计带来的第一个根本性优势是责任边界绝对清晰。在 LLM-centric 架构里“谁该负责校验格式”“谁该决定是否重试”“谁该记录错误日志”这些问题的答案往往是模糊的可能混在 prompt 里可能藏在 tool call 的参数里可能由模型内部的 reasoning chain 暗中决策。而在 LangGraph 的 state-first 架构里答案只有一个state schema 定义了契约router 函数定义了规则node 函数只负责读写 state。没有灰色地带没有隐式约定。这直接降低了协作成本——前端同学只需要关心TriageState里哪些字段要传、哪些是可选后端同学只维护validate_node的校验逻辑运维同学盯着app.get_graph().draw_mermaid_png()生成的流程图就能一眼看出瓶颈在哪。我见过太多团队因为“模型该不该做校验”吵得不可开交最后发现大家对“校验”这个词的理解都不同有人指正则匹配有人指数据库查重有人指调用风控 API。而 state schema 用Optional[str]和Literal[refund, replacement]这样的类型注解把歧义直接焊死。2.2 “状态即契约”为什么共享状态比函数调用更健壮LangGraph 的核心哲学是把传统编程中“函数 A 调用函数 BB 返回结果给 A”的链式思维彻底扭转为“所有节点共同维护一个全局状态对象各自读取、各自更新”的协作模式。这听起来像分布式系统的共享内存但它的精妙之处在于状态不是数据容器而是执行契约。我们来看TriageState的定义class TriageState(TypedDict, totalFalse): order_id: Optional[str] issue_type: Optional[IssueType] resolution_preference: Optional[Resolution] is_valid: bool errors: list[str] result: Optional[str]这个看似简单的 TypedDict实际承载了三层契约存在性契约order_id是Optional[str]意味着它允许为空但一旦出现就必须是字符串。如果某个 node 错误地赋值为int(123)Python 的类型检查器或运行时的 Pydantic 验证会立刻报错而不是让错误潜伏到下游节点崩溃。这比在每个函数入口写assert isinstance(order_id, str)干净一百倍。责任契约is_valid: bool和errors: list[str]这两个字段明确划定了“验证节点”的职责边界。validate_node的唯一任务就是根据当前 state 中的order_id、issue_type等字段计算出is_valid的布尔值并把所有校验失败的原因塞进errors。它不负责重试不负责路由不负责生成最终文案——这些统统交给 router 和其他节点。这种单一职责让单元测试变得极其简单你只需 mock 一个{order_id: invalid}的 state断言validate_node返回的is_valid为Falseerrors包含特定字符串即可。演进契约result: Optional[str]是整个工作流的“终点站”。只有当is_valid为True且所有分支refund/replacement都完成自己的逻辑后finalize_node才能往这里写入最终结果。这个字段的存在天然形成了一个“完成态”的标志位。你可以轻松实现这样的监控逻辑if final_state.get(result) and not final_state.get(errors): send_to_slack(✅ 订单分流成功)。而如果用传统函数链你需要在每个分支末尾都手动拼接一个 success flag极易遗漏。我曾经重构过一个电商客服机器人旧系统用的是 Chain-of-Thought prompt每次对话都要把历史消息、用户意图、商品信息全塞进 context window。结果一遇到长对话token 就爆模型就开始胡编乱造。迁移到 LangGraph 后我们把conversation_history、current_intent、resolved_items全部拆成 state 字段。最直观的变化是当用户突然问“刚才说的优惠券怎么领”旧系统要重新解析整个长文本找线索新系统直接state[resolved_items][-1][coupon_code]就拿到——状态不是累赘是索引是记忆是让系统“记得自己做过什么”的基础设施。2.3 “路由即决策”为什么把 if/else 提到图层是架构升级的关键在传统代码里if/else是嵌在函数内部的控制流语句。比如def handle_order_issue(data): if not validate_inputs(data): return ask_for_missing_info(data) elif data[resolution_preference] refund: return process_refund(data) else: return process_replacement(data)这种写法的问题在于决策逻辑和业务逻辑耦合在一起无法独立测试、无法动态替换、无法可视化追踪。当你想加一个“如果用户是 VIP优先走人工审核”分支时你得改handle_order_issue函数还得确保所有调用它的上游都兼容新逻辑。LangGraph 的add_conditional_edges把这个困境彻底解开。route_after_validation函数只做一件事基于当前 state返回一个字符串标签。它不执行任何业务操作不修改 state不调用外部服务。它的输出只是告诉 graph“下一步请去input节点” 或 “请去refund节点”。这个标签就是 graph 的“交通信号灯”。这种分离带来的好处是颠覆性的可测试性爆炸提升你可以为route_after_validation写一组穷举测试# 测试用例1验证失败 → 重试 assert route_after_validation({is_valid: False}) retry_input # 测试用例2验证通过且偏好退款 → 走退款流 assert route_after_validation({is_valid: True, resolution_preference: refund}) refund_flow # 测试用例3验证通过但偏好未设置 → 默认走换货还是报错由你定义 assert route_after_validation({is_valid: True, resolution_preference: None}) replacement_flow这些测试跑起来毫秒级覆盖了所有决策分支而无需启动整个工作流、mock 数据库、等待 LLM 响应。可观测性质变当工作流跑起来你不仅能拿到最终result还能拿到完整的configurable追踪日志看到每一步的state快照和next_node标签。比如日志里显示[Step 1] Node: input → State: {order_id: ORD-123} → Next: validate [Step 2] Node: validate → State: {order_id: ORD-123, is_valid: False, errors: [issue_type missing]} → Next: retry_input [Step 3] Node: input → State: {order_id: ORD-123, issue_type: damaged} → Next: validate这比任何 LLM 的 reasoning trace 都更真实、更可靠。你不需要猜模型“为什么选了这条路”因为next_node是route_after_validation函数明确返回的而它的输入state是你完全掌控的。策略热替换成为可能想象一下业务方突然要求“下周起所有late类型的订单无论用户选什么都强制走退款”。在传统架构里这可能需要改process_refund和process_replacement两个函数还要改handle_order_issue的判断逻辑。在 LangGraph 里你只需要重写route_after_validationdef route_after_validation(state: TriageState) - str: if not state.get(is_valid, False): return retry_input # 新增策略late订单强制退款 if state.get(issue_type) late: return refund_flow if state.get(resolution_preference) refund: return refund_flow return replacement_flow编译新 graph上线完成。业务逻辑refund_node,replacement_node一行不动。这就是“决策与执行分离”赋予的敏捷性。3. 实操细节解析从状态定义到图编译每一步的底层逻辑3.1 状态 Schema 的工程化设计不只是类型更是业务语义TriageState看似简单但它的每一个字段选择都经过了反复权衡。我们来逐行拆解其背后的设计考量class TriageState(TypedDict, totalFalse): order_id: Optional[str] # ← 为什么是 Optional[str]而不是 str issue_type: Optional[IssueType] # ← 为什么用 Literal 而不是 str resolution_preference: Optional[Resolution] # ← 同上且为何不直接写 refund|replacement is_valid: bool # ← 为什么是 bool而不是 enum errors: list[str] # ← 为什么是 list[str]而不是 str result: Optional[str] # ← 为什么是 Optional[str]而不是 dictorder_id: Optional[str]这里的Optional不是偷懒而是精确表达业务语义——“订单号是用户可选填写的字段系统允许其为空”。如果定义为str那么在input_node刚启动、还没收到任何输入时state[order_id]就会触发KeyError。而Optional[str]配合TypedDict的totalFalse意味着这个字段可以完全不存在于 state 字典中。input_node的职责就是检查state里有没有order_id没有就尝试从外部如 HTTP 请求体获取并写入有就跳过。这种“按需填充”的模式让节点职责更轻量也避免了初始化一堆空字符串的冗余代码。issue_type: Optional[IssueType]IssueType Literal[damaged, late, wrong_item]这个定义是强类型安全的基石。它比str有三大优势IDE 自动补全你在写state[issue_type] ...时PyCharm 会直接弹出三个选项杜绝拼写错误比如damged这种低级错误。静态类型检查mypy能在编译期就报错error: Incompatible types in assignment (expression has type str, variable has type Literal[damaged, late, wrong_item])而不是等运行时报KeyError。业务约束显性化Literal明确告诉所有协作者“这个问题类型只有这三种不可能有第四种”。这比在文档里写“请确保 issue_type 是 damanged/late/wrong_item 之一”有力得多。当产品提出要加一个missing_tracking类型时你必须显式修改IssueType这个动作本身就会触发一次代码审查确保所有人知晓变更。is_valid: bool为什么不用enum因为bool是最简、最无歧义的状态表示。“有效”或“无效”没有第三种状态。enum会引入不必要的复杂度比如ValidationStatus.VALID,ValidationStatus.INVALID,ValidationStatus.PENDING而PENDING在这个场景下毫无意义——验证节点执行完结果必然是True或False。用boolroute_after_validation的判断逻辑就干净利落if not state.get(is_valid):没有歧义没有例外。errors: list[str]这里用list[str]而非str是为了一次性收集所有错误。想象用户同时漏填issue_type和resolution_preference如果errors是单个字符串你只能存一个错误信息另一个就丢了。而list[str]允许validate_node这样写errors [] if not state.get(issue_type): errors.append(issue_type is required) if not state.get(resolution_preference): errors.append(resolution_preference is required) if not re.match(r^ORD-\d$, state.get(order_id, )): errors.append(order_id must match format ORD-12345) return {is_valid: len(errors) 0, errors: errors}这种“批量校验、批量反馈”的模式极大提升了用户体验——用户一次就能看到所有要补的信息而不是填完一个再提示下一个。result: Optional[str]为什么是str而非dict因为在这个最小可行示例中result的唯一目的是作为工作流的“出口标识”。它不承载结构化数据只是一段人类可读的摘要用于快速验证流程是否走通。Refund flow selected for ORD-12345. Issue type: late. Next steps: verify eligibility → initiate refund → notify customer.这样的字符串足够让开发者一眼确认order_id解析正确、issue_type被识别、refund_flow被选中、后续步骤已规划。如果你需要返回结构化数据比如退款金额、预计到账时间那应该定义新的 state 字段如refund_amount: Optional[float]、estimated_settlement_date: Optional[str]而不是把所有东西都塞进result字符串里。result是“总结”不是“数据库”。提示TypedDict的totalFalse参数至关重要。它表示这个字典的键是“部分可选”的而不是“所有键都必须存在”。这完美契合了工作流的渐进式状态演化初始 state 可能只有order_id中间 state 会加上issue_type和errors最终 state 才会有result。如果设为totalTrue那么从第一步开始你就得把所有字段都初始化为None或默认值代码会变得臃肿且违背直觉。3.2 节点函数的编写规范纯函数、无副作用、只读写 stateLangGraph 的节点Node函数必须严格遵守“纯函数”原则给定相同的输入 state永远返回相同的输出 state 更新且不产生任何外部副作用如调用 API、写数据库、发邮件。这是保证工作流可预测、可重放、可测试的铁律。以validate_node为例一个符合规范的实现如下import re from typing import Dict, Any def validate_node(state: Dict[str, Any]) - Dict[str, Any]: 纯校验函数只读取 state 中的字段计算 is_valid 和 errors不修改任何外部状态。 order_id state.get(order_id, ) issue_type state.get(issue_type) resolution_preference state.get(resolution_preference) errors [] # 校验 order_id 格式 if not isinstance(order_id, str) or not re.match(r^ORD-\d$, order_id): errors.append(order_id must be a string matching format ORD-12345) # 校验 issue_type 是否在允许范围内 valid_issue_types {damaged, late, wrong_item} if issue_type not in valid_issue_types: errors.append(fissue_type must be one of {valid_issue_types}, got {issue_type}) # 校验 resolution_preference 是否在允许范围内 valid_preferences {refund, replacement} if resolution_preference not in valid_preferences: errors.append(fresolution_preference must be one of {valid_preferences}, got {resolution_preference}) # 检查是否所有必填字段都存在且有效 is_valid len(errors) 0 and all([ isinstance(order_id, str) and re.match(r^ORD-\d$, order_id), issue_type in valid_issue_types, resolution_preference in valid_preferences ]) return { is_valid: is_valid, errors: errors }这个函数的每一行都在践行纯函数原则无外部依赖没有requests.get()没有db.query()没有logger.info()。所有逻辑都基于输入的state字典。无状态修改它不修改传入的state对象Python 中 dict 是可变对象但这里我们只读不state.pop()或state.update()而是返回一个全新的字典只包含需要更新的字段。LangGraph 会自动将这个返回字典 merge 到当前 state 中。确定性输出只要state相同errors列表的内容和is_valid的值就绝对相同。你可以用pytest写一百个测试用例它们都会稳定通过。对比一个“坏”的实现# ❌ 危险违反纯函数原则 def bad_validate_node(state: Dict[str, Any]) - Dict[str, Any]: # 1. 外部副作用调用风控API risk_score requests.post(https://api.risk.com/check, json{order_id: state[order_id]}).json()[score] if risk_score 0.8: logger.warning(fHigh risk order {state[order_id]} detected) # 2. 外部副作用打日志 # 3. 修改传入的 state危险 state[is_high_risk] True # 4. 逻辑混乱混合了校验和业务决策 if risk_score 0.95: return {is_valid: False, errors: [Order blocked by risk system]} # ... 后续校验逻辑这个函数的问题是灾难性的它依赖外部 API导致测试必须 mock它打日志污染了纯计算逻辑它直接修改state可能导致并发问题它把风控决策业务逻辑和基础校验技术逻辑混在一起无法单独测试校验功能。在生产环境中一旦风控 API 慢或挂了整个工作流就会卡死或超时。注意纯函数不等于“不能做任何事”。validate_node完全可以做复杂的计算比如调用本地的机器学习模型.pkl文件加载后预测、执行 SQL 查询如果数据库连接是本地的、无网络 IO、解析 PDF使用pypdf。关键在于这些操作必须是确定性的、无外部网络依赖的、不改变全局状态的。如果必须调用外部 API正确的做法是把这个调用封装成一个独立的 node如call_risk_api_node并在 graph 中明确将其放在validate_node之后、route_after_validation之前。这样validate_node依然保持纯净而外部依赖被隔离在专属节点里便于 mock 和监控。3.3 图构建的底层机制add_conditional_edges如何驱动状态机workflow.add_conditional_edges(validate, route_after_validation, {...})这行代码是整个工作流的“心脏起搏器”。它的执行逻辑远比表面看起来的if/else复杂。我们来深挖 LangGraph 内部是如何解析并执行这条指令的注册路由函数当你调用add_conditional_edges时LangGraph 并不会立即执行route_after_validation。它只是把这个函数对象连同你提供的{retry_input: input, ...}映射字典一起注册到validate节点的“条件边集合”中。节点执行与状态捕获当工作流运行到validate节点时LangGraph 会调用validate_node(state)得到一个Dict例如{is_valid: False, errors: [...]}。将这个返回字典merge到当前state中形成新的state此时state[is_valid]已更新。关键一步LangGraph 会将这个更新后的完整 state作为参数传递给route_after_validation函数。路由函数执行与标签提取route_after_validation函数被调用它只读取state中的字段如state[is_valid]然后返回一个字符串标签如retry_input。标签匹配与边选择LangGraph 拿到这个返回的字符串标签retry_input去查找你注册的映射字典{retry_input: input, refund_flow: refund, ...}。它找到键retry_input对应的值input。执行跳转LangGraph 决定下一步应该执行input节点。它会将当前state已包含validate_node的更新作为输入调用input_node(state)。这个过程揭示了一个重要事实route_after_validation的输入是validate_node执行完毕后的最新 state而不是validate_node执行前的 state。这意味着路由决策是基于“最新事实”的而不是基于“猜测”。validate_node可以在更新is_valid的同时也更新errors、warnings、甚至suggested_fix等辅助字段route_after_validation都能读到。实操中我经常利用这一点做“智能重试”。比如在validate_node里如果发现order_id格式错误但看起来像ORD12345少了-我可以这样写# 在 validate_node 中 if not re.match(r^ORD-\d$, order_id): # 尝试自动修复 fixed_order_id re.sub(r^ORD(\d)$, rORD-\1, order_id) if re.match(r^ORD-\d$, fixed_order_id): # 修复成功记录建议 suggested_fix fDid you mean {fixed_order_id}? errors.append(forder_id format invalid. {suggested_fix}) # 同时把修复后的 ID 也存入 state供重试时使用 return {is_valid: False, errors: errors, suggested_order_id: fixed_order_id}然后在input_node里我可以检查state.get(suggested_order_id)如果存在就直接用它填充order_id字段而不是让用户重新输入。这就是“状态驱动”的力量validate_node发现问题并提供解决方案input_node执行解决方案route_after_validation只负责判断“现在够不够格往下走”。提示add_conditional_edges的第三个参数映射字典中键key必须是route_after_validation函数可能返回的所有字符串值的超集。如果你的路由函数有时返回vip_review但映射字典里没有vip_review这个 keyLangGraph 会抛出KeyError。因此强烈建议在路由函数的末尾加一个else分支返回一个兜底标签如default并在映射字典中明确处理它避免流程意外中断。4. 完整实操流程从零开始搭建、运行、调试一个无模型工作流4.1 环境准备与依赖安装避开版本陷阱在开始编码前环境配置是成败的第一关。LangGraph 的版本迭代较快不同版本的 API 有细微差别。根据我实测截至 2024 年 Q3最稳定、文档最全的组合是Python: 3.10 或 3.11LangGraph 对 3.12 的支持尚不完善LangGraph:langgraph0.1.50这是目前StateGraphAPI 最成熟的版本0.2.x系列引入了CompiledGraph等新概念但文档和社区案例较少其他依赖:typing-extensions4.5.0TypedDict的高级特性需要、graphviz用于生成流程图安装命令# 创建并激活虚拟环境强烈推荐避免包冲突 python -m venv langgraph-env source langgraph-env/bin/activate # Linux/Mac # langgraph-env\Scripts\activate # Windows # 安装核心依赖 pip install langgraph0.1.50 typing-extensions # 如果要生成 PNG 流程图还需安装 graphviz系统级 # Ubuntu/Debian: sudo apt-get install graphviz libgraphviz-dev # Mac (Homebrew): brew install graphviz # Windows: 下载 Graphviz 安装包 (https://graphviz.org/download/)并把 bin 目录加入 PATH pip install pygraphviz # Python 绑定注意不要用pip install langgraph不带版本因为最新版0.2.x的StateGraph初始化方式已改为StateGraph(TriageState).add_node(...)而本文基于0.1.x的StateGraph(TriageState)。版本不匹配会导致AttributeError: StateGraph object has no attribute add_node这类错误。我踩过这个坑花了两小时才定位到是版本问题。4.2 从零编写完整的可运行代码与逐行注释下面是你可以直接复制、粘贴、运行的完整代码。我将每一行都做了详细注释解释其作用和背后的工程考量。# -*- coding: utf-8 -*- 无模型智能体工作流订单问题分流系统 作者一位在生产环境摔过无数跟头的工程师 说明此代码可在 Python 3.10 环境中直接运行无需任何 LLM 或 API Key。 # 1. 导入必要模块 from typing import TypedDict, Optional, Literal, Dict, Any, List from langgraph.graph import StateGraph, END # LangGraph 的核心类 import re # 用于正则校验 # 2. 定义业务类型Literal 确保类型安全 IssueType Literal[damaged, late, wrong_item] Resolution Literal[refund, replacement] # 3. 定义状态 Schema核心契约 class TriageState(TypedDict, totalFalse): TriageState: 订单问题分流工作流的共享状态。 totalFalse 表示字典的键是可选的符合工作流渐进式填充的特性。 order_id: Optional[str] # 订单号字符串可为空 issue_type: Optional[IssueType] # 问题类型必须是预定义的三种之一 resolution_preference: Optional[Resolution] # 解决偏好必须是预定义的两种之一 is_valid: bool # 校验结果True 表示所有字段有效且完整 errors: List[str] # 校验错误列表用于收集所有问题 result: Optional[str] # 最终结果摘要仅用于验证流程 # 4. 定义节点函数input_node收集输入 def input_node(state: Dict[str, Any]) - Dict[str, Any]: input_node: 模拟从外部如 HTTP 请求、消息队列获取用户输入。 此处为简化我们模拟一个“智能补全”逻辑 - 如果 state 中已有 order_id且格式接近正确如 ORD12345则尝试修复。 - 否则保持原样等待后续校验。 注意这是一个纯函数不调用任何外部服务。 # 获取当前 state 中的 order_id current_order_id state.get(order_id, ) # 如果 order_id 存在但格式错误缺少 -尝试自动修复 if isinstance(current_order_id, str) and re.match(r^ORD\d$, current_order_id): # 修复在 ORD 和数字间插入 - fixed_order_id re.sub(r^ORD(\d)$, rORD-\1, current_order_id) print(f[input_node] 自动修复 order_id: {current_order_id} - {fixed_order_id}) return {order_id: fixed_order_id} # 如果 order_id 为空或格式正确不作任何修改 # 返回空字典表示 state 无需更新 return {} # 5. 定义节点函数validate_node核心校验 def validate_node(state: Dict[str, Any]) - Dict[str, Any]: validate_node: 对 state 中的关键字段进行严格校验。 这是工作流的“质量门禁”必须确保所有业务规则在此处被检查。 order_id state.get(order_id, ) issue_type state.get(issue_type) resolution_preference state.get(resolution_preference) errors [] # 校验 order_id必须是非空字符串且匹配 ORD-12345 格式 if not isinstance(order_id, str) or not order_id.strip(): errors.append(order_id cannot be empty) elif not re.match(r^ORD-\d$, order_id.strip()): errors.append(order_id must match format ORD-12345 (e.g., ORD-12345)) # 校验 issue_type必须是预定义的三种之一 valid_issue_types {damaged, late, wrong_item} if issue_type not in valid_issue_types: errors.append(fissue_type must be one of {list(valid_issue_types)}, got {issue_type}) # 校验 resolution_preference