大家好，今天我们将创建一个智能客服机器人，它能够记录用户的食物订单到真实数据库中，并允许用户查看他们的订单。这是一个相对高级的Langgraph项目，大家可以先看一下前面介绍的Langgraph的基础课程。
项目概述
我们要构建的系统是一个模拟登录的用户，可以创建订单或查看订单，其他功能则被屏蔽。首先我们会用一个虚拟令牌设置用户，并通过系统消息给机器人一个身份。然后我们会分类处理用户的意图：
-
离题意图：完全屏蔽。
-
获取所有订单：查询数据库，按用户过滤并返回订单。
-
创建订单：首先验证请求，缺失信息则提示用户补充。如果信息完整，将时间格式化为有效的时间戳，然后创建订单并写入数据库，最后给用户一个友好的响应。
这整个过程分步骤进行，下面我们详细介绍实现过程。
数据库设置
由于我们使用真实数据库，第一步是设置数据库。需要Docker支持。我们使用docker-compose文件设置一个PostgreSQL数据库，并在5433端口运行。运行docker-compose up命令启动数据库，可以通过docker ps确认数据库是否运行。
用户和产品表
我们用SQLAlchemy创建用户和订单表。首先加载API密钥，并安装SQLAlchemy：
pip install SQLAlchemy
我们创建一个基础类Base，并继承自declarative_base，然后创建用户表、订单表和食物项表。用户表和订单表通过关系关联。我们在食物项表中提供一些示例食物，如比萨等等。
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, sessionmaker
Base = declarative_base()
class Customer(Base):
__tablename__ = 'customers'
id = Column(Integer, primary_key=True)
name = Column(String)
orders = relationship("Order", back_populates="customer")
class Order(Base):
__tablename__ = 'orders'
id = Column(Integer, primary_key=True)
customer_id = Column(Integer, ForeignKey('customers.id'))
food_item_id = Column(Integer, ForeignKey('food_items.id'))
delivery_address = Column(String)
order_date = Column(String)
customer = relationship("Customer", back_populates="orders")
food_item = relationship("FoodItem")
class FoodItem(Base):
__tablename__ = 'food_items'
id = Column(Integer, primary_key=True)
name = Column(String)创建表后，我们插入一个示例用户和一些食物项。
engine = create_engine('postgresql+psycopg2://user:password@localhost:5433/mydatabase')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
# 创建示例用户和食物项
new_customer = Customer(name="John")
session.add(new_customer)
session.commit()
pizza = FoodItem(name="Pizza")
salami = FoodItem(name="Salami")
session.add_all([pizza, salami])
session.commit()代理状态和验证逻辑
为了在节点间传递数据，我们需要一个代理状态AgentState。这个状态包含用户问题、系统消息、用户消息、客户名、工具调用、生成的最终输出等信息。我们通过一系列函数对用户请求进行检查和处理。
class AgentState:
def __init__(self):
self.question = ""
self.messages = []
self.customer_name = ""
self.tool_calls = []
self.order_check = ""
self.generation = ""
self.system_message = ""验证用户请求的示例如下：
def validate_order(question):
# 假设这是一个调用LLM的函数，返回验证结果
if "food items" not in question or "delivery address" not in question or "order date" not in question:
return False
return True工具调用和函数
我们需要一些工具函数来处理数据库查询和订单创建。使用LangChain框架中的工具装饰器来定义这些函数。
@tool
def create_order(customer_name, food_items, delivery_address, order_date):
session = Session()
customer = session.query(Customer).filter_by(name=customer_name).first()
if not customer:
return "Customer not found"
food_item = session.query(FoodItem).filter_by(name=food_items).first()
if not food_item:
return "Food item not found"
new_order = Order(customer_id=customer.id, food_item_id=food_item.id, delivery_address=delivery_address, order_date=order_date)
session.add(new_order)
session.commit()
return "Order placed successfully"
@tool
def get_all_orders(customer_name):
session = Session()
customer = session.query(Customer).filter_by(name=customer_name).first()
if not customer:
return "Customer not found"
orders = session.query(Order).filter_by(customer_id=customer.id).all()
if not orders:
return "No orders found"
return [{"order_id": order.id, "food_item": order.food_item.name, "delivery_address": order.delivery_address, "order_date": order.order_date} for order in orders]系统消息和用户消息
接下来我们需要创建系统消息和用户消息，并通过LangChain框架绑定工具。
system_message = "You are an order bot. You can help users create and view food orders."
def create_human_message(question):
return {"role": "user", "content": question}
def create_system_message():
return {"role": "system", "content": system_message}意图识别和路由
通过LangChain框架，我们可以识别用户意图，并根据意图进行相应的处理。
def identify_intent(question):
if "order" in question:
return "create_order"
elif "view" in question or "see" in question:
return "get_all_orders"
else:
return "off_topic"
def route_intent(intent, state):
if intent == "off_topic":
return "I can only help you with creating or viewing orders."
elif intent == "create_order":
if validate_order(state.question):
return create_order(state.customer_name, state.order_check["food_items"], state.order_check["delivery_address"], state.order_check["order_date"])
else:
return "Your order information is incomplete."
elif intent == "get_all_orders":
return get_all_orders(state.customer_name)生成最终消息
最后一步是生成最终消息，告知用户订单创建的结果或显示用户的所有订单。
def generate_final_message(state):
if state.intent == "create_order":
if validate_order(state.question):
return create_order(state.customer_name, state.order_check["food_items"], state.order_check["delivery_address"], state.order_check["order_date"])
else:
return "Your order information is incomplete."
elif state.intent == "get_all_orders":
return get_all_orders(state.customer_name)
else:
return "I can only help you with creating or viewing orders."测试与运行
我们可以通过一系列测试来验证系统的功能是否正常。
state = AgentState()
state.customer_name = "John"
state.question = "我需要预定一个披萨，地址是北京路123号4栋506，时间为明天上午10点"
# 更新系统消息和用户消息
state.messages.append(create_system_message())
state.messages.append(create_human_message(state.question))
# 识别意图并路由
state.intent = identify_intent(state.question)
response = route_intent(state.intent, state)
print(response)通过以上步骤，我们成功创建了一个智能客服机器人，它能够处理用户的食物订单请求并返回相应的结果。这是一个复杂而强大的系统，展示了现代科技在客户服务领域的巨大潜力。
关注我，每天带你开发一个AI应用，每周二四六直播，欢迎多多交流。
