langGraph从入门到精通（七）——基于 LangGraph 的结构化数据AI 代理自动入库实战

基于 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.5

4.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% 准确。

欢迎评论区留言讨论核心主题相关的问题~