1
import os
2
from langchain_openai import ChatOpenAI
3

4
os.environ["OPENAI_API_KEY"] = "sk-xxx" # 输入自己的api-key
5

6
llm = ChatOpenAI(
7
  base_url="https://api.deepseek.com/v1",
8
  model="deepseek-chat",
9
  temperature=0
10
)

1
def chat_with_model(state):
2
  """generate structured output"""
3
  messages = state['messages']
4
  structured_llm = llm.with_structured_output(FinalResponse, method="function_calling")
5
  response = structured_llm.invoke(messages)
6
  return {"messages": [response]}

1
from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData, text
2
from sqlalchemy.orm import declarative_base, sessionmaker
3

4

5
# 创建基类
6
Base = declarative_base()
7

8
# 定义 UserInfo 模型
9
class User(Base):
10
  __tablename__ = 'users'
11
  id = Column(Integer, primary_key=True)
12
  name = Column(String(50))
13
  age = Column(Integer)
14
  email = Column(String(100))
15
  phone = Column(String(15))
16

17
# 数据库连接配置
18
# 格式：mysql+pymysql://用户名:密码@主机:端口/数据库名?charset=utf8mb4
19
# 本地连接说明：
20
# - 用户名：你的 MySQL 用户名（默认通常是 root）
21
# - 密码：你的 MySQL 密码
22
# - 主机：localhost 或 127.0.0.1（本地连接）
23
# - 端口：3306（MySQL 默认端口，如果使用默认端口可以省略）
24
# - 数据库名：要连接的数据库名称
25
# - charset=utf8mb4：设置字符集为 utf8mb4，支持更广泛的 Unicode 字符
26

27
# MySQL 连接信息（不包含数据库名，用于创建数据库）
28
MYSQL_USER = 'root'
29
MYSQL_PASSWORD = '123456'
30
MYSQL_HOST = 'localhost'
31
MYSQL_PORT = '3306'
32
DATABASE_NAME = 'langgraph_agent'
33

34
# 先连接到 MySQL 服务器（不指定数据库），用于创建数据库
35
server_uri = f'mysql+pymysql://{MYSQL_USER}:{MYSQL_PASSWORD}@{MYSQL_HOST}:{MYSQL_PORT}?charset=utf8mb4'
36
server_engine = create_engine(server_uri, echo=False)
37

38
# 创建数据库（如果不存在）
39
with server_engine.connect() as conn:
40
    conn.execute(text(f"CREATE DATABASE IF NOT EXISTS `{DATABASE_NAME}` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"))
41
    conn.commit()
42

43
# 连接到指定的数据库
44
DATABASE_URI = f'mysql+pymysql://{MYSQL_USER}:{MYSQL_PASSWORD}@{MYSQL_HOST}:{MYSQL_PORT}/{DATABASE_NAME}?charset=utf8mb4'
45
engine = create_engine(DATABASE_URI, echo=True)
46

47
# 如果表不存在，则创建表
48
Base.metadata.create_all(engine)
49

50
# 创建会话
51
Session = sessionmaker(bind=engine)
52
session = Session()

1
def final_answer(state):
2
  """generate natural language responses"""
3
  messages = state['messages'][-1]
4
  response = messages.final_output.response
5
  return {"messages": [response]}
6

7
def insert_db(state):
8
  """Insert user information into the database"""
9
  session = Session() # 确保为每次操作创建新的会话
10
  try:
11
    result = state['messages'][-1]
12
    output = result.final_output
13
    user = User(name=output.name, age=output.age, email=output.email, phone=output.phone)
14
    session.add(user)
15
    session.commit()
16
    return {"messages": [f"数据已成功存储至Mysql数据库。"]}
17
  except Exception as e:
18
    session.rollback()  # 出错时回滚
19
    return {"messages": [f"数据存储失败，错误原因：{e}"]}
20
  finally:
21
    session.close()  # 关闭会话

1
from langchain_core.messages import AnyMessage
2
from typing_extensions import TypedDict, Annotated
3
import operator
4

5
class AgentState(TypedDict):
6
  messages: Annotated[list[AnyMessage], operator.add]
7

8
def generate_branch(state: AgentState):
9
  result = state['messages'][-1]
10
  output = result.final_output
11
  if isinstance(output, UserInfo):
12
    return True
13
  elif isinstance(output, ConversationalResponse):
14
    return False

1
from langgraph.graph import StateGraph
2

3
graph = StateGraph(AgentState)
4

5
graph.add_node("chat_with_model", chat_with_model)
6
graph.add_node("final_answer", final_answer)
7
graph.add_node("insert_db", insert_db)
8

9
graph.set_entry_point("chat_with_model")
10

11
graph.add_conditional_edges(
12
  "chat_with_model",
13
  generate_branch,
14
  {True: "insert_db", False: "final_answer"}
15
)
16

17
graph.set_finish_point("final_answer")
18
graph.set_finish_point("insert_db")
19

20
graph = graph.compile()

1
from langchain_core.messages import HumanMessage
2

3
query = "我叫Hygen，今年18岁，邮箱地址是ciallo@gmial.com，电话是01000721"
4
input_message = {"messages": [HumanMessage(content=query)]}
5
result = graph.invoke(input_message)
6
print(result)

1
{'messages': [HumanMessage(content='我叫Hygen，今年18岁，邮箱地址是ciallo@gmial.com，电话是01000721', additional_kwargs={}, response_metadata={}), FinalResponse(final_output=UserInfo(name='Hygen', age=18, email='ciallo@gmial.com', phone='01000721')), '数据已成功存储至Mysql数据库。']}

1
query = "你好，请你介绍一下你自己"
2
input_message = {"messages": [HumanMessage(content=query)]}
3
result = graph.invoke(input_message)
4
print(result)

1
{'messages': [HumanMessage(content='你好，请你介绍一下你自己', additional_kwargs={}, response_metadata={}), FinalResponse(final_output=ConversationalResponse(response='你好！我是一个AI助手，专门设计来帮助用户处理各种任务和回答问题。我可以协助您进行信息查询、文本处理、数据分析等
2
多种工作。如果您有任何需要帮助的地方，请随时告诉我，我会尽力为您提供支持！')), '你好！我是一个AI助手，专门设计来帮助用户处理各种任务和回答问题。我可以协助您进行信息查询、文本处理、数据分析等多种工作。如果您有任何需要帮助的地方，请随时告诉我，我会尽力为您提供支持！']}