找回密码
立即注册
搜索
热搜: Java Python Linux Go
发回帖 发新帖

3623

积分

0

好友

481

主题
发表于 2 小时前 | 查看: 4| 回复: 0

在 Agent 开发框架层出不穷的当下,如果要手搓一个自己的 AI 助手,了解它们背后的设计思路就成了向最聪明的那批人“偷师”的捷径。我最近在研究 LangGraph 时这种感觉尤其强烈。虽然对很多生活中的小项目,用它有点像“杀鸡用牛刀”,但其中蕴含的架构思想,比如 State、Node、Edge 的解耦设计,很有启发性。

在 AI 能力爆炸的今天,“语法”反而不重要了。新框架、新语言迭代太快,个体根本追不上社区的进化速度。但技术的进化与生物进化相似,再复杂的变异,其起点都是几种基础模块。LangGraph 中的基础模块,正是理解复杂 Agent 逻辑的关键。

上一篇文章我们剖析了 State 的概念,本文接着拆解剩下的两大要素:Nodes 和 Edges,并会通过一个“邮件起草 Bot”实例,演示如何将人类反馈融入工作流。

本文核心内容:

  • Nodes 的定义与设计原则
  • Normal Edges 与 Conditional Edges
  • 如何利用 Pydantic 实现 LLM 结构化输出
  • 实战:构建带人类反馈的“邮件起草 Bot”

01. 什么是 Node?

如果把 Graph 看作一个智能工厂,State 就是贯穿全局的“共享记事本”,所有部门(Node)都可以读写它。而 Node,就是流水线上那些处理 State 的“机械臂”或“加工站点”——它接收 State,加工数据,再把结果更新回去。

一只机械臂手绘图,象征 LangGraph 中的 Node 概念

那么,在实际开发中,什么时候应该新建一个 Node 呢?遵循以下原则可以帮你避免陷入纠结:

  • 触发更新时:只要你想对 State 做出改动,就需要新建一个 Node。
  • 坚持单一职责:就像写代码追求可复用、可维护一样,Node 的功能越纯粹,未来用基础模块搭建复杂逻辑就越轻松。如果一个 Node 看上去很“精分”,一会儿关注 A,一会儿关注 B,那就该拆分了。
  • 调用 LLM 看人设:如果调用大模型,尽量给每个 Node 分配“单一人设”,清晰定义它在整个链条中的角色。

在代码层面,Node 的本质是一个函数(实际是 Runnable)。得益于 LangGraph 底层封装,我们只需在 Node 函数中输出 State 需要更新的“增量部分”,完全不用把整个 State 重新抄写一遍。

假设我们定义了这样的 State,其中 messagecount 使用 operator.add 累加,memberTier 则直接覆盖:

代码截图:learningState 类定义,展示 message、count 和 memberTier 字段

对应的 Node 函数可以写成这样,它只返回 messagecount 两个 Key 的增量数据:

代码截图:sayHello 函数定义,返回增量的 message 和 count

假如初始化 State 长这样:

代码截图:sharedBlackBoard1 初始化,message 列表含 I am Mia

经过 sayHello 处理后,message 会自动新增一条,count 也会加 1:

代码截图:处理后的 State,message 列表新增了 Hello!

这里有个关键点:如果 Node 的输出 Key 与 State 不重叠,它就只是“路过”,不会改变 State。此外,Graph 中有两个默认的特殊节点——STARTEND,它们作为常量存在,用来标记工作流的“起点”与“终点”。为什么需要它们?因为执行图时数据是有序流动的,没有 START 就无法定义顺序,没有 END 底层搜索循环就停不下来。

代码截图:从 langgraph 导入 START 和 END

一个简单流程图:START 直连 sayHello 再到 END

02. 什么是 Edge?

Edge 就好比连接各个“机械臂”的“传送带”,在底层,它可以被理解为一个调度字典,决定物料下一步流向哪里。

一幅手绘风格的双机械臂与传送带插图,象征 Edge 概念

LangGraph 中有两种基础边:

  • 普通边 (Normal Edges):固定映射,逻辑是“如果 A 结束,永远去 B”。字典的 Key 是上游节点名,Value 是下游节点名。这里全部使用字符串名而非函数名,是为了解耦,方便随时重命名或替换节点。
  • 条件边 (Conditional Edges):动态映射,依据 Router 函数的输出来分流。

下图模拟了 Edge 结构,其中 edges 是普通边,conditionalEdges 是条件边:

代码截图:模拟 Edge 内部结构的字典示例

定义普通边的语法很简单,就是显式调用 add_edge 并传入字符串名字:

代码截图:调用 myGraph.add_node 和 add_edge 方法

03. 条件边:让 Agent 学会“看人下菜碟”

条件边需要三要素:上游节点的“名字”、路由函数、以及一张路径映射表。它本质是让 Graph 具备了判断能力。

手绘风格的带分支传送带示意图

比如我们要做一个 VIP 用户分流:用户是 VIP 就去 VIP 通道,否则去普通通道。流程大致如下:

流程图:从 sayHello 分叉出 VIPPath 和 ordinaryPath

首先定义 Router 函数 checkVIP,它根据 State 里的 memberTier 字段返回不同的“意图字符串”:

代码截图:checkVIP 函数定义,判断是否为 VIP

调用 add_conditional_edges 时的语法如下,注意 pathMap 的作用是把意图字符串映射到真实的 Node 名字:

代码截图:add_conditional_edges 的完整调用示例

当然,如果你的 Router 函数返回的字符串本身就和 Node 名字一致,也可以直接传一个列表,框架会自动做“恒等映射”。官网示例经常这么做:

代码截图:add_conditional_edges 使用列表的简写形式

04. 组装 Graph 与编译

理解了三大要素,就可以组装 Graph 了。步骤概括为:传 State、加 Node、连 Edge。下面是一段完整的构建代码示例:

代码截图:完整的 Graph 构建代码,包含多个节点和条件边

定义好图后,务必执行一次 compile()。这步操作会解析那些字符串形式的配置,将 Annotated 的归并函数等还原为真实对象,为实际运行做好准备。

代码截图:调用 compile 方法

编译完成后,调用 invoke 传入初始 State,工作流就能跑起来了:

代码截图:调用 invoke 并打印结果

主程序逻辑与 Graph 本身要分清楚。最简单的主程序就是定义初态、调用 Graph、输出结果:

代码截图:从初始化 State 到 invoke 并打印结果的完整代码

05. 实战:做个会反思的“邮件起草 Bot”

理论讲完,我们来动手做一个能跟人对话、反复修改的邮件起草助手。整个 Graph 只有一个核心 Node,但包含了与 LLM 交互和融合人类反馈的完整逻辑。

第一步:定义 State

起草邮件需要知道消息历史、分类、草稿和用户确认状态。因此 State 设这四个字段,isConfirmed 默认 False 以驱动修改循环。

代码截图:myState 类定义,包含 messages、category、draft、isConfirmed

第二步:用 Pydantic 约束输出

LLM 的输出不可控,必须用 Pydantic 给它装个“精密收纳盒”。它基于 Python 类型提示,比普通字典约束更强:数据不对会直接报错,还能自动做类型强制与序列化。在 AI 工程里,Pydantic 就是连接“天马行空的 LLM”与“严谨逻辑”的桥梁。

我们定义 ProposerOutput 类作为输出格式:

代码截图:ProposerOutput 类定义,继承 BaseModel,含 description

description 非常重要,这不仅是给开发者看的,更会作为提示传给 LLM,帮它理解输出要求。

第三步:系统提示词

提示词定义了这个 Node 的人设,建议抽离到函数外部方便微调:

代码截图:邮箱起草助手的系统提示词

第四步:编写核心 Node

为了让例子可交互(只要基于终端就能跑),我们在 Node 内部用 input() 来接收人类反馈。这里必须强调:生产环境千万别这么干!因为 input() 会阻塞线程,导致资源无法释放。正确的做法我们下篇文章讨论。

人类输入通过 HumanMessage 封装后推入 State 的 messages 列表。

代码截图:emailDraft 函数开头,使用 input 获取人类消息

接下来是核心:如何让 LLM 返回结构化数据?重点在于 with_structured_output

先用自己熟悉的 API(这里以 Gemini 为例)初始化模型,然后把 ProposerOutput 传给 with_structured_output 方法。LangGraph 在底层会玩一个“骗术”:它把我们的数据结构伪装成一个 Function Tool,让 LLM 误以为自己在调工具,从而严格按参数格式输出。如果 LLM 输出的 JSON 校验不通过,Pydantic 会直接报错,并不让错误数据蒙混过关。

代码截图:导入 ChatGoogleGenerativeAI 并初始化 myLLM

代码截图:调用 with_structured_output 方法

构建消息列表时,通常会把系统提示词和 State 里的历史消息拼接起来。为简单起见,这里把历史消息全量传给 LLM(实际生产需用摘要或截断来防止超 Token 限制)。

代码截图:构建 final_messages 列表并调用 structured_llm.invoke

拿到结构化结果后,我们在终端打印出草稿:

代码截图:打印邮件草稿的代码片段

最后,Node 返回“增量”数据。为了让 LLM 记住历史,必须把它的回答(AIMessage)也追加到 messages 字段。

代码截图:emailDraft 函数的完整 return 语句

第五步:定义条件边与循环

这个 Graph 需要循环,条件边的 Router 函数根据 isConfirmed 决定是结束还是再跑一遍 emailDraft

代码截图:checkIsConfirmed 路由函数,判断 isConfirmed 状态

第六步:构建与运行

把节点和边拼装起来,编译,并用 display 查看生成的流程图,能直观看到那个从 emailDraft 指回自身的循环回路。

代码截图:emailBuilder 图的完整构建、编译与流程图展示代码

流程图:emailDraft 与 START、END 之间的循环连接

主程序用一个 while 循环不断触发 Graph,直到用户确认为止。

代码截图:main 函数内的 while 循环,不断 invoke 图

最终的运行结果:系统发来问候,我输入“写一个请假申请”后,Bot 马上生成了一份带占位符的模板。当我补充个人信息后,它草拟出了完整的请假邮件。我再提出“礼貌一点”,它立刻润色得更加得体。

运行截图:助手生成的第一版请假申请草稿

运行截图:用户补充信息后助手生成的第二版草稿

运行截图:用户要求礼貌一点后助手润色完成的终版草稿

06. 总结

总的来说,Node 是加工 State 的工作站,我们遵循“单一职责”并只返回增量数据;Edge 是连接它们的传送带,分普通和条件两种,依赖“字符串名字”进行解耦导航;Graph 需要编译后通过 invoke 运行。

而借助 Pydantic 与 with_structured_output 这套组合拳,我们能驯服 LLM 的野性输出,将其转变成严丝合缝的 Python 对象。

虽然直接在 Node 里用 input() 有悖最佳实践,但它确实清晰地展示了人类反馈如何注入 Agent 循环。下一篇我们聊聊怎么用“打断(Interrupt)”机制来优雅地解决阻塞问题,让生产级的交互真正流畅起来。




上一篇:Hadoop 3.3.6 HA高可用部署实战:基于HDFS Federation的进阶优化
下一篇:反测绘攻防实战:协议伪装与端口蜜罐欺骗Nmap扫描
您需要登录后才可以回帖 登录 | 立即注册

手机版|小黑屋|网站地图|云栈社区 ( 苏ICP备2022046150号-2 )

GMT+8, 2026-6-7 21:25 , Processed in 0.725700 second(s), 39 queries , Gzip On.

Powered by Discuz! X3.5

© 2025-2026 云栈社区.

快速回复 返回顶部 返回列表