ARTICLE DETAIL

资讯详情

深耕编程入门与网站建设的一线实战洞察。

AI Agent需求分析:从模糊意图到可执行节点的硬核拆解

AI Agent需求分析:从模糊意图到可执行节点的硬核拆解 1. 这不是写PRD是给AI Agent“画经络图”——为什么90%的Agent项目死在需求分析这一步我带过7个从零启动的AI Agent落地项目最常被问的问题不是“怎么选模型”而是“需求文档写了20页为什么一跑就崩”。上周刚帮一家做智能客服的团队重构他们的Agent架构他们交来的《需求说明书》里写着“用户提问后系统应准确理解意图并给出专业回答”这种描述放在十年前写Java Web应用还勉强能用放到Agent工程里等于让外科医生只带一把剪刀进手术室——工具和任务完全不匹配。AI Agent的需求分析本质不是梳理功能点而是解构人类决策链路在机器世界里的映射关系。你写的每一条需求都得能翻译成可执行的节点、可传递的状态、可验证的边界条件。比如“用户问‘我的订单为什么还没发货’系统要查物流、看库存、联系仓库”这背后藏着至少4个隐性约束物流API的超时阈值必须≤800ms否则用户会重复提问、库存状态缓存更新频率不能低于5分钟否则出现幻读、仓库接口失败时需自动降级到人工坐席队列否则形成服务黑洞、所有环节必须支持异步重试且重试次数≤3次否则雪崩。这些根本不会出现在传统PRD里但缺一个Agent就会在真实流量下集体失智。真正卡住大多数人的是分不清“业务语言”和“Agent语言”的转换成本。业务方说“要智能”工程师想“加个RAG”而实际需要的是定义状态机里多少个节点、每个节点的输入输出Schema长什么样、失败时状态如何迁移、重试策略绑定在哪个层级、监控指标该埋在哪条边……这些全得在需求分析阶段拍板。LangGraph官方文档里那张著名的“State Graph”示意图底下小字写着“State must be serializable, immutable, and versioned”——这句话就是需求分析的宪法。我见过太多团队把state设计成嵌套字典结果调试时发现某个节点修改了state里深层的list导致上游节点拿到脏数据也见过把retry逻辑写在tool调用层结果workflow重启时重试计数器归零连续触发三次风控拦截。所以别再用Word写需求了。你现在手里的不是功能清单是一份Agent神经网络的布线图。它得标清楚哪些信号走高速通道同步调用哪些走慢速总线消息队列哪些信号必须带校验码checksum哪些线路要预留冗余带宽fallback路径。接下来我会用一个真实电商售后Agent的拆解案例带你把“用户要查订单状态”这种模糊需求变成LangGraph里可运行的17个节点、3个状态分支、5种异常处理策略——不是教你怎么写代码而是教你用工程师的显微镜看清需求背后的物理约束。2. 需求拆解四步法从模糊意图到可执行节点的硬核转换2.1 第一步剥离“伪需求”锁定真正的决策断点很多团队一上来就列功能“支持语音输入”、“接入微信小程序”、“生成PDF报告”。这些全是技术方案不是需求。真正的Agent需求只有一种当系统收到某个输入时在什么条件下必须做出什么确定性响应。我们拿电商售后场景里最典型的“用户问‘我的订单为什么还没发货’”来练手。先做减法❌ “支持多轮对话” → 这是交互方式不是决策逻辑❌ “响应时间2秒” → 这是SLA属于性能需求不在架构设计层解决✅ “当订单状态为‘已支付’且创建时间48小时必须触发物流查询” → 这才是决策断点输入订单状态时间→ 条件48h→ 动作触发查询我习惯用“如果…那么…”句式暴力拆解。把原始需求逐句改写原始需求“用户能自助查询订单物流信息”改写1“如果用户提供了订单号那么系统必须校验订单号格式有效性”改写2“如果订单号校验通过那么必须查询订单中心获取基础状态”改写3“如果订单状态为‘已支付’且当前时间-创建时间48h那么必须调用物流API”改写4“如果物流API返回‘无单号’那么必须检查是否为虚拟商品并跳过物流环节”你会发现真正需要架构设计的是改写3和改写4里的条件分支。其他都是标准组件正则校验、HTTP客户端。这就是为什么LangGraph强调“state-driven”因为每个if条件最终都会变成state schema里的一个字段比如order_status: str、payment_time: datetime、is_virtual_product: bool。提示每次写完一个“如果…那么…”立刻问自己这个条件能否用Python的if语句直接判断如果答案是“需要查数据库才能知道”说明你漏掉了前置节点——数据库查询本身就得是一个独立节点。2.2 第二步构建状态树State Tree而不是功能树传统需求分析画功能模块图Agent需求分析必须画状态迁移树。以售后Agent为例我们定义初始state包含class OrderState(TypedDict): order_id: str user_id: str order_status: Literal[created, paid, shipped, delivered] payment_time: Optional[datetime] logistics_info: Optional[dict] retry_count: int fallback_to_human: bool注意这里没有user_question字段——因为问题文本应该在进入第一个节点前就被LLM解析成结构化意图否则后续所有节点都要处理NLU违背单一职责原则。我们强制要求第一个节点parse_intent的输出必须是{ intent: check_logistics, order_id: ORD123456, confidence: 0.92 }状态树的根节点是parse_intent它的子节点分叉依据是intent值check_logistics→ 进入物流查询分支cancel_order→ 进入取消订单分支request_refund→ 进入退款分支每个分支再往下拆。比如物流分支parse_intent └── check_logistics ├── validate_order_id # 校验订单号格式 │ └── (valid) → get_order_status │ └── (invalid) → return_error ├── get_order_status # 查订单中心 │ └── (status paid and time 48h) → call_logistics_api │ └── (status shipped) → return_logistics_info └── call_logistics_api # 调物流API └── (success) → enrich_logistics └── (timeout) → increment_retry fallback_to_human关键洞察状态树的每个叶子节点都对应一个不可再分的原子操作。call_logistics_api不能包含“重试逻辑”因为重试是状态迁移的决策不是API调用本身。我们把重试计数器放在state里由call_logistics_api节点执行后由下一个handle_api_result节点根据retry_count和错误类型决定是否重试。2.3 第三步标注所有“暗流”——那些不写在需求里却决定生死的约束客户说“要快”没说“快到什么程度不影响准确性”。但Agent架构里速度和精度永远在博弈。我们给电商售后Agent标出5类暗流约束约束类型具体表现架构影响我踩过的坑时序约束物流API平均响应800msP99达2.3s必须设置timeout3s且重试间隔≥1s避免压垮对方曾用timeout1s导致30%请求被截断用户看到“系统繁忙”数据一致性订单中心状态更新延迟≤5s物流API延迟≤30sget_order_status和call_logistics_api必须串行不能并发查并发导致看到“已支付”却查不到物流单号用户以为系统故障降级策略当物流API失败率5%自动切到本地缓存需在state里存last_success_time节点check_fallback_condition实时计算失败率缓存未设TTL用户看到3天前的物流信息安全边界单个用户1分钟内最多触发5次物流查询在state里加query_timestamps: List[datetime]validate_rate_limit节点校验用Redis计数器但没考虑分布式锁导致限流失效可观测性每个节点执行耗时、重试次数、错误码必须上报所有节点函数签名强制加tracer: Tracer参数统一埋点最初只埋了成功日志线上故障时无法定位卡在哪这些约束决定了节点间的连接方式。比如check_fallback_condition不能直接连call_logistics_api中间必须插一个update_state_with_metrics节点来刷新失败计数器——因为状态更新必须发生在决策之前。2.4 第四步用“失败路径”反向验证需求完整性正向写需求容易漏掉异常流。我的方法是对每个主流程强制写出3条失败路径并检查是否有节点能处理它。主流程用户问物流 → 解析意图 → 校验订单号 → 查订单状态 → 调物流API → 返回结果失败路径1物流API返回{code: 503, msg: Service Unavailable}应该由handle_api_result节点捕获判断retry_count 3→ 更新state重试计数器 → 发送send(call_logistics_api, state)否则 → 设置fallback_to_humanTrue→ 跳转到人工入口失败路径2订单中心返回{status: cancelled}get_order_status节点输出order_statuscancelled下游call_logistics_api节点收到后直接返回{logistics: 订单已取消无需发货}注意这里不能让call_logistics_api去判断订单状态它只管调API失败路径3用户连续3次输错订单号validate_order_id节点每次失败都记录invalid_attempts 1当invalid_attempts 3时state里设block_userTrue下一个节点check_user_block拦截请求返回友好提示注意所有失败路径的终点必须是明确的state变更或外部动作如发短信、写DB。禁止出现“记录日志后忽略”这种设计——Agent没有“忽略”这个选项它必须对每个输入给出确定性响应。3. LangGraph实战拆解把需求文档变成可运行的Graph3.1 State Schema设计为什么必须用TypedDict而不是dict很多人图省事用普通dict定义state# ❌ 危险类型不安全IDE无法提示调试时字段名拼错不报错 state { order_id: ORD123, retry_count: 0, logistics_data: None }LangGraph要求state必须可序列化且不可变TypedDict是唯一选择from typing import TypedDict, Optional, Literal from datetime import datetime class OrderState(TypedDict): order_id: str user_id: str order_status: Literal[created, paid, shipped, delivered, cancelled] payment_time: Optional[datetime] logistics_info: Optional[dict] retry_count: int fallback_to_human: bool invalid_attempts: int block_user: bool为什么这么麻烦三个血泪教训字段名拼写错误logistics_info写成logistics_infoo运行时才报KeyError而TypedDict在IDE里直接标红类型混淆retry_count本该是int但某次API返回字符串3后续retry_count 1报错TypedDict强制类型检查缺失字段新增block_user字段后旧state加载时没这个keyTypedDict在state.get(block_user, False)时IDE会警告“可能None”更关键的是LangGraph的add_edge和add_conditional_edges依赖state字段做路由判断。比如def should_call_logistics(state: OrderState) - str: if state[order_status] paid and \ state[payment_time] and \ (datetime.now() - state[payment_time]).total_seconds() 48*3600: return call_logistics_api else: return return_no_action如果payment_time是Optional[datetime]IDE能提示你加is not None判断如果是普通dict你可能忘了判空线上直接500。3.2 节点函数编写每个函数必须是纯函数且有明确副作用LangGraph节点函数有铁律输入state输出state不修改原state副作用只能是send或yield。看一个典型错误写法# ❌ 绝对禁止修改了原state对象 def call_logistics_api(state: OrderState): api_response requests.post(https://logistics.com/api, json{order_id: state[order_id]}) state[logistics_info] api_response.json() # 错直接改原对象 return state正确写法必须深拷贝LangGraph内部会处理但你得保证函数纯净# ✅ 正确返回新state原state不变 def call_logistics_api(state: OrderState) - OrderState: try: response requests.post( https://logistics.com/api, json{order_id: state[order_id]}, timeout3 ) response.raise_for_status() # 创建新state只更新必要字段 return { **state, logistics_info: response.json(), retry_count: state[retry_count] # 重试计数器在此节点不增加 } except requests.Timeout: # 超时不算失败重试由下游节点控制 return {**state} except Exception as e: # 其他错误计入重试 return { **state, retry_count: state[retry_count] 1 }为什么强调纯函数因为LangGraph可能对同一state多次调用同一节点比如重试时如果节点有副作用会导致状态污染。我们曾遇到一个bugget_order_status节点里写了logger.info(fQuerying order {state[order_id]})结果重试时日志打两遍运维误判为双倍流量。3.3 边缘连接Edges设计条件路由的3种写法与陷阱LangGraph的add_conditional_edges是需求落地的核心。电商Agent里最关键的路由是should_proceed_to_logisticsdef should_proceed_to_logistics(state: OrderState) - str: # 错误示范用字符串比较易出错 if state[order_status] paid: return call_logistics_api # 正确写法用Literal枚举IDE自动补全 if state[order_status] is OrderStatus.PAID: return call_logistics_api # 更健壮组合多个条件 if (state[order_status] in [OrderStatus.PAID, OrderStatus.SHIPPED]) and \ state[payment_time] and \ (datetime.now() - state[payment_time]).total_seconds() 48*3600: return call_logistics_api return return_no_action三种边缘写法对比写法适用场景风险点我的选择add_edge(node_a, node_b)固定顺序无条件跳转节点间耦合度高难以插入监控节点仅用于start → parse_intent这种必经路径add_conditional_edges(node_a, { path1: node_b, path2: node_c })二元/多元分支条件简单字符串键名易拼错分支逻辑分散主力写法但键名用Enum代替字符串add_edge(node_a, END, conditionlambda s: s[fallback_to_human])短路退出如降级到人工条件函数难调试建议封装成独立函数用于紧急逃生通道如风控拦截特别注意条件函数必须幂等。我们曾写过一个条件函数# ❌ 危险每次调用都生成新时间戳导致路由结果不稳定 def is_timeout_expired(state): return (datetime.now() - state[start_time]).total_seconds() 30结果Agent在重试时反复在两个节点间震荡。正确做法是把超时判断移到节点里state里存timeout_deadline: datetime。3.4 异常处理机制LangGraph里没有try-catch只有状态迁移传统编程用try-catch捕获异常Agent架构里异常必须转化为state变更。LangGraph提供两种机制机制1节点内处理推荐def handle_api_result(state: OrderState) - OrderState: if state[logistics_info] is None: # API失败更新重试计数 new_retry state[retry_count] 1 if new_retry 3: return {**state, fallback_to_human: True} else: return {**state, retry_count: new_retry} else: return state机制2使用interrupt中断慎用# 当检测到恶意刷单行为时立即中断流程 if state[invalid_attempts] 5: raise NodeInterrupt(User blocked for abuse)中断会暂停graph执行需要外部系统如客服后台人工介入。我们只在风控场景用日常异常都走状态迁移。实操心得所有节点函数顶部加统一错误兜底def robust_node(state: OrderState) - OrderState: try: return actual_logic(state) except Exception as e: logger.error(fNode failed: {e}, extra{state: state}) # 返回安全state避免流程中断 return {**state, error_message: str(e), fallback_to_human: True}4. 需求分析交付物一份能让开发、测试、产品都看懂的Agent蓝图4.1 可视化状态图用draw.io画出比代码更清晰的架构别用PlantUML用draw.io画状态图。关键要素节点形状圆角矩形普通节点、菱形条件判断、六边形外部系统调用、云朵人工介入连线标注箭头旁写条件如order_statuspaid time48h颜色编码绿色成功路径红色失败路径黄色降级路径状态字段在节点下方用小字列出该节点读/写的state字段如READ: order_id, order_status / WRITE: logistics_info我们给电商Agent画的状态图里call_logistics_api节点下方标注READ: order_id, retry_count WRITE: logistics_info, retry_count SIDE EFFECT: HTTP POST to logistics.com这张图交付给测试团队后他们直接导出为测试用例表测试场景输入state期望输出state验证点物流API超时{order_id:ORD1,retry_count:0}{retry_count:0,logistics_info:null}不增加retry_count第三次重试失败{order_id:ORD1,retry_count:2}{retry_count:3,fallback_to_human:True}fallback标志置为True4.2 节点契约文档Node Contract每个节点的“宪法”这是需求分析最核心的交付物格式固定【节点名】validate_order_id 【职责】校验订单号格式是否符合正则^[A-Z]{3}\d{6}$并检查是否存在于订单库 【输入state字段】order_id 【输出state字段】is_valid_order: bool, error_message: str, invalid_attempts: int 【副作用】查询订单库SELECT 1 FROM orders WHERE id ? 【失败处理】若DB查询失败返回is_valid_orderFalseerror_message系统繁忙 【性能要求】P99响应时间≤100ms为什么叫“契约”因为开发、测试、运维都按这个契约工作开发者实现时必须满足所有输入输出约束测试者写case时只关心契约里写的字段运维监控时只看契约里定义的性能指标我们曾用这份契约发现一个致命问题get_order_status节点契约写的是“查询订单中心”但开发者实际调用了缓存服务导致状态延迟。契约里明确写了“必须调用订单中心API”问题当场暴露。4.3 边界测试用例集覆盖所有“不应该发生但一定会发生”的场景需求分析最后一步是列出10个最恶心的测试用例它们不来自需求文档而来自生产环境用户输入订单号ORD123但订单中心返回{status:deleted}软删除状态物流API返回{code:200,data:null}空响应同一订单号1秒内被5个不同用户查询缓存击穿payment_time字段为None订单未支付order_id包含SQL注入字符 OR 11网络抖动导致call_logistics_api节点执行一半被killretry_count达到3后用户又发来新问题state未重置时钟不同步服务器时间比订单中心快2小时logistics_info字段JSON过大1MB超出LangGraph默认序列化限制用户连续发送100条“查物流”触发限流但没返回友好提示这些用例驱动开发第9条让我们在call_logistics_api节点加了JSON大小校验第6条促使我们给所有节点加retry(stopstop_after_attempt(3))装饰器第10条直接催生了rate_limiter专用节点。注意每个用例必须对应到具体节点。比如“订单中心返回deleted”对应get_order_status节点的契约而不是笼统说“整个系统”。5. 避坑指南那些只有亲手踩过才知道的Agent需求分析雷区5.1 雷区1把LLM当万能胶忽视状态设计的物理限制新手最爱写“让LLM自己决定下一步做什么”。这相当于让司机自己画地图。LangGraph的StateGraph要求每个节点职责单一而LLM节点天然适合做意图解析和结果合成不适合做状态迁移决策。我们曾有个项目让LLM输出下一步节点名# ❌ 灾难设计 llm_output llm.invoke(fBased on state {state}, which node to call next?) next_node llm_output.content.strip() # 可能是call_logistics_api或check_stock结果上线后发现LLM偶尔输出CALL_LOGISTICS_API大写而graph里节点名是call_logistics_api路由失败LLM在压力下输出I think we should call logistics根本不是节点名没法做单元测试因为LLM输出不可预测正确解法LLM只负责结构化路由逻辑写死在Python里。LLM输出必须是严格schema{ next_action: check_logistics, confidence: 0.95, reason: 用户明确询问物流信息 }然后用确定性函数路由def route_by_llm_output(state: OrderState) - str: if state[llm_output][next_action] check_logistics: return validate_order_id elif state[llm_output][next_action] cancel_order: return check_cancel_eligibility5.2 雷区2忽略state版本演进导致老state无法加载Agent上线后需求会变比如新增is_virtual_product字段。如果直接修改TypedDict# v1 class OrderState(TypedDict): order_id: str # v2错误破坏兼容性 class OrderState(TypedDict): order_id: str is_virtual_product: bool # 新增字段老state只有order_id加载时会报错。正确做法是所有字段Optional并提供迁移函数class OrderState(TypedDict): order_id: str is_virtual_product: Optional[bool] # 显式声明可选 def migrate_state_v1_to_v2(old_state: dict) - OrderState: v1 state - v2 state return { order_id: old_state[order_id], is_virtual_product: old_state.get(is_virtual_product, False) }我们在电商Agent里维护了state_migrations.py每个版本升级前用脚本批量迁移历史state。上线前必须跑通所有迁移路径测试。5.3 雷区3在节点里做“不该做的决策”导致状态污染常见错误在call_logistics_api节点里既调API又判断是否要重试# ❌ 错误混合了IO和决策逻辑 def call_logistics_api(state: OrderState): response requests.post(...) if response.status_code 503: # 这里做重试决策但state还没更新retry_count return send(call_logistics_api, state) # 无限循环正确分层call_logistics_api只负责IO返回原始responsehandle_api_result纯决策节点根据response和state决定下一步increment_retry纯state更新节点只改retry_count这样每个节点都可单独测试handle_api_result的单元测试可以mock任何responsedef test_handle_503_error(): state {retry_count: 2, order_id: ORD1} result handle_api_result({ **state, api_response: {status_code: 503} }) assert result[retry_count] 3 assert result[fallback_to_human] is False5.4 雷区4用自然语言写需求却用代码思维验收产品经理写“用户满意度要提升20%”。这根本不是Agent需求而是业务目标。对应的Agent需求应该是“当用户问题包含‘不满意’、‘差评’、‘投诉’等关键词时必须在300ms内转接人工坐席”“人工坐席接手前必须把完整对话历史、订单信息、用户画像摘要打包发送”我们要求所有需求必须可测量❌ “响应更快” → ✅ “P95响应时间≤1.2s”❌ “更准确” → ✅ “意图识别准确率≥92%测试集”❌ “用户体验好” → ✅ “单次对话人工介入率≤8%”验收时测试团队直接跑自动化脚本对比指标是否达标。没有模糊空间。5.5 雷区5忘记Agent的“记忆”是有成本的LangGraph的state存在内存里越大越慢。曾有个团队把整段对话历史存进state# ❌ 危险state爆炸 state { conversation_history: [ {role: user, content: 你好}, {role: assistant, content: 您好请问有什么可以帮您}, ... ] }结果10轮对话后state超2MB序列化耗时飙升。正确做法对话历史用外部向量库存储state里只存conversation_id敏感信息如用户手机号必须脱敏后再进state大文件如图片base64绝对不进state用临时URL代替我们在电商Agent里规定state总大小≤128KB超过自动触发告警。上线后监控显示99%的state在45KB以内。6. 工程实践结语需求分析不是起点而是Agent生命的DNA写完这份需求拆解文档我打开终端运行langgraph-cli graph visualize看着自动生成的状态图在浏览器里展开——17个节点像神经元一样连接3条主路径如血管般延伸5个红色的fallback节点像安全阀一样分布在关键位置。这不是一张设计图而是Agent的基因图谱。当第一个用户问“我的订单为什么还没发货”时这个图谱会指挥数据流经parse_intent、validate_order_id、get_order_status……最终在return_logistics_info节点组装出那句“您的订单已于今天10:23发货预计明天送达”。需求分析之所以痛苦是因为它强迫你直面AI的物理现实LLM不是神它是有延迟的APIstate不是无限内存它是有大小限制的容器重试不是魔法它是需要计数器和超时的机械过程。那些在Word里写满“智能”、“高效”、“人性化”的需求文档本质上是在逃避这些物理约束。我最后想分享一个真实案例去年帮一家银行做信用卡逾期催收Agent客户最初的需求是“用AI温和地提醒用户还款”。我们花了两周时间把这句话拆解成温和 语气词过滤器禁用“必须”、“立即”等词情绪识别节点检测用户愤怒值0.7时降级提醒 3种触达渠道APP推送、短信、外呼的优先级矩阵基于用户历史响应率动态调整还款 12种还款方案最低还款、分期、延期的资格校验规则每条规则对应一个独立节点最终交付的不是“温和提醒”而是一个有237个状态分支、41个决策节点、7种降级策略的精密系统。上线后逾期回收率提升18%但更重要的是它从未出现过一次“温和过头”导致用户错过还款也从未“强硬过头”引发投诉——因为每一个“温和”的定义都刻在了state schema里每一次“提醒”的时机都写在了条件路由中。所以别再问“Agent项目怎么拆”问问自己你敢不敢把“用户要查订单状态”这句话拆成17个节点、3个状态分支、5种异常处理敢的话你已经站在了Agent工程化的门口。
返回列表