基于 LangGraph 的结构化数据AI 代理自动入库实战
1 导语
在企业级 AI 应用中，仅能“聊天”的 Agent 远远不够，如何将对话中的关键信息自动识别并精准持久化到业务数据库，是实现业务闭环的关键。本文将带你通过一个亲测有效的实战案例，掌握利用LangGraph的条件边（Conditional Edges）与SQLAlchemyORM 框架，构建一个能够自动识别用户信息并将其存储至 MySQL 数据库的智能代理系统。
2 技术栈清单
- Python== 3.11.14
- langgraph== 1.0.5
- langchain-core== 1.2.7
- SQLAlchemy== 2.0.x
- PyMySQL== 1.1.x
- MySQL== 8.0/5.7
3 项目核心原理
本项目核心在于利用 LangGraph 的StateGraph进行工作流编排。系统通过 LLM 的Structured Output（结构化输出）能力将非结构化对话转化为 Pydantic 模型对象。随后，通过Conditional Edges（条件边）进行逻辑分流：若提取到有效用户信息，则流转至数据库操作节点；否则执行普通对话回复。
4 实战步骤
4.1 环境准备
首先需要配置数据库连接信息，并安装必要的数据库驱动与 ORM 框架。
# 安装数据库连接相关依赖pipinstallsqlalchemy==2.0.31pymysql==1.1.1langgraph==1.0.54.2 代码实现
4.2.1 数据库模型与会话配置
使用 SQLAlchemy 定义User模型，映射至数据库中的users表。
fromsqlalchemyimportcreate_engine,Column,Integer,Stringfromsqlalchemy.ormimportdeclarative_base,sessionmaker Base=declarative_base()classUser(Base):# 定义 ORM 模型__tablename__='users'id=Column(Integer,primary_key=True)name=Column(String(50))age=Column(Integer)email=Column(String(100))phone=Column(String(15))# 数据库连接 URI (需替换为实际配置)DATABASE_URI='mysql+pymysql://user:pass@localhost:3306/db'engine=create_engine(DATABASE_URI)Session=sessionmaker(bind=engine)4.2.2 核心节点逻辑
编写数据库插入节点，利用Session将 AI 提取的数据写入。
definsert_db(state):# 数据库存储节点session=Session()try:output=state['messages'][-1].final_output# 获取上一个节点的结构化输出user=User(name=output.name,age=output.age,email=output.email,phone=output.phone)session.add(user)# 添加记录session.commit()# 提交事务return{"messages":["数据已成功存储至数据库。"]}exceptExceptionase:session.rollback()# 异常回滚return{"messages":[f"存储失败：{e}"]}finally:session.close()# 关闭会 fancy 话4.2.3 图编排与条件路由
通过add_conditional_edges实现根据 AI 输出类型自动选择路径。
# 构建状态图graph=StateGraph(AgentState)graph.add_node("chat_with_model",chat_with_model)graph.add_node("insert_db",insert_db)graph.add_conditional_edges("chat_with_model",generate_branch,# 路由决策函数{True:"insert_db",False:"final_answer"}# 路径映射)4.3 功能测试
输入包含个人信息的语句，验证系统是否能够自动识别并完成入库。
query="我叫奥特曼，今年38岁，邮箱地址是aoteman@qq.com"result=graph.invoke({"messages":[HumanMessage(content=query)]})5 核心代码解析
5.1 条件边路由逻辑解析
划重点：这是实现复杂业务逻辑的核心。
defgenerate_branch(state):output=state['messages'][-1].final_outputreturnisinstance(output,UserInfo)- 核心作用：作为图的“指挥官”，判断 LLM 的输出是属于用户信息还是普通聊天。
- 为何这样实现：通过
isinstance检查对象类型，实现了业务逻辑的硬编码控制与 AI 生成能力的完美解耦。
5.2 SQLAlchemy 会话管理设计
session=Session()# 确保为每次操作创建新的会话- 核心作用：保证数据库连接的线程安全性。
- 关键参数：
Session配合try...except...finally结构，确保了即使在分布式环境下，数据库事务也能正确提交或回滚，避免连接泄露。
6 效果验证
执行后，控制台会输出echo=True的 SQL 语句，同时数据库users表中会新增对应记录。
7 踩坑记录
7.1 DATABASE_URI 格式错误
- 错误现象：
NoSuchModuleError: Can't load plugin: sqlalchemy.dialects.mysql.pymysql - 根因分析：未安装
pymysql驱动或连接串拼写错误。 - 解决方案：亲测有效的连接串格式为
mysql+pymysql://user:pass@host:port/db?charset=utf8mb4。
7.2 Session 提交后未关闭
- 错误现象：数据库连接池溢出，后续检测任务挂起。
- 根因分析：在节点函数中打开了 Session 但未在
finally中关闭。 - 解决方案：严格执行“打开-使用-关闭”的生命周期管理。
7.3 Structured Output 类型判断失效
- 错误现象：总是跳转到
final_answer节点。 - 根因分析：模型输出的 Pydantic 对象与
generate_branch中判断的类型不一致。 - 解决方案：检查
with_structured_output绑定的类定义，确保路由函数识别准确。
8 总结与扩展
本文通过LangGraph + SQLAlchemy实现了从语义理解到持久化存储的全自动化。核心收获在于理解了如何通过条件边来管理 Agent 的执行路径。未来我们可以进一步扩展，在insert_db节点之前加入人工确认节点（Human-in-the-loop），确保入库数据的 100% 准确。
欢迎评论区留言讨论核心主题相关的问题~
