1. 案例目标
本案例旨在构建一个基于智能体的动态槽位填充系统，实现智能对话系统，能够分析用户请求并自动收集必要信息，通过对话补充缺失信息。
系统主要实现以下目标：
- 实现动态槽位填充功能，自动识别并收集必要信息
- 支持多种任务类型，包括餐厅预订、会议安排、酒店预订和航班预订
- 通过自然对话收集缺失信息，提供流畅的用户体验
- 自动验证输入信息的有效性
- 展示LangGraph框架在构建复杂对话系统中的应用
2. 技术栈与核心依赖
LangChain
LangGraph
OpenAI GPT
Python
Jupyter Notebook
3. 环境配置
3.1 安装依赖
# 安装所需包 %pip install langchain-opentutorial from langchain_opentutorial import package package.install( [ "langsmith", "langchain", "langchain_core", "langchain_openai", "langgraph", ], verbose=False, upgrade=False, )3.2 环境变量设置
# 设置环境变量 from langchain_opentutorial import set_env set_env( { "OPENAI_API_KEY": "", "LANGCHAIN_API_KEY": "", "LANGCHAIN_TRACING_V2": "true", "LANGCHAIN_ENDPOINT": "https://api.smith.langchain.com", "LANGCHAIN_PROJECT": "Agent-Based-Dynamic-Slot-Filling", } )4. 案例实现
4.1 系统架构设计
4.2 常量和提示模板定义
4.2.1 任务槽位定义
TASK_SLOTS = { "restaurant": { "restaurant_address": "餐厅地址/位置(城市名称)", "number_of_people": "人数(数字)", "reservation_datetime": "预订日期和时间(YYYY/MM/DD HH:MM格式)", }, "meeting": { "meeting_datetime": "会议日期和时间(YYYY/MM/DD HH:MM格式)", "platform": "视频会议平台(zoom/teams/google meet)", "meeting_duration": "会议时长(分钟)", }, "hotel": { "hotel_location": "酒店位置(城市名称)", "check_in_date": "入住日期(YYYY/MM/DD)", "check_out_date": "退房日期(YYYY/MM/DD)", "room_type": "房间类型(single/double/suite)", "number_of_guests": "客人数量(数字)", }, "flight": { "departure_city": "出发城市", "arrival_city": "到达城市", "departure_date": "出发日期(YYYY/MM/DD HH:MM格式)", "return_date": "返程日期(YYYY/MM/DD HH:MM格式)", "passenger_count": "乘客数量(数字)", "seat_class": "座位等级(economy/business/first)", }, }4.2.2 任务规则定义
TASK_RULES = { "restaurant": """ 1. 不要询问已经提供的信息 2. 首先确认具体的餐厅位置(城市名称) 3. 人数只接受数字值 4. 预订日期和时间接受YYYY/MM/DD HH:MM格式""", "hotel": """ 1. 不要询问已经提供的信息 2. 首先确认酒店位置(城市名称) 3. 入住/退房日期接受YYYY/MM/DD格式 4. 让用户从single/double/suite中选择房间类型 5. 客人数量只接受数字值""", "meeting": """ 1. 不要询问已经提供的信息 2. 首先确认会议室位置 3. 参会人数只接受数字值 4. 会议日期和时间接受YYYY/MM/DD HH:MM格式 5. 会议时长接受小时数(例如, 1小时, 2小时)""", "flight": """ 1. 不要询问已经提供的信息 2. 首先确认出发和到达位置 3. 出发和返程日期接受YYYY/MM/DD HH:MM格式 4. 乘客数量只接受数字值""", }4.2.3 提示模板定义
# 任务分类模板 TASK_CLASSIFICATION_TEMPLATE = """请分析用户消息并选择适当的预订/预约任务。 可用任务列表: - restaurant: 餐厅预订 - meeting: 视频会议预订 - hotel: 酒店预订 - flight: 航班预订 用户消息: {user_message} 请按以下格式回复: {{"task": "task_name", "confidence": 0.XX}} confidence应该是0.0到1.0之间的值，表示意图分类的确定性。 如果无法确定意图，将confidence设置为0。 """ # 槽位提取模板 SLOT_EXTRACTION_TEMPLATE = """请从以下对话中提取与{task_type}预订相关的信息。 所需信息: {required_slots} 当前槽位状态: {slots} 对话: {messages} 最后一条消息: {last_message} 当前日期: {current_date} 请严格遵循以下规则: 1. 日期和时间转换规则: - 所有日期必须为YYYY/MM/DD HH:MM格式 - 如果只提到"下周"，询问具体日期和时间(保持为null) - 只有在指定星期几时才转换为日期(例如,"下周一") - 如果未指定时间，保持为null 2. 不完整日期/时间情况: - 仅"下周" → null - 仅"晚上" → null - "下周一晚上" → YYYY/MM/DD 19:00 - "明天午餐" → YYYY/MM/DD 12:00 3. 数字必须转换为数字格式(例如,"四个人" → "4") 4. 位置名称保持原样(例如,"曼哈顿","纽约") 5. 将不确定的信息标记为null 请按以下JSON格式回复提取的信息: {{"restaurant_address": "位置或null", "number_of_people": "数字或null", "reservation_datetime": "YYYY/MM/DD HH:MM或null"}} """ # 响应模板 RESPONSE_TEMPLATE = """以友好的语气继续对话，同时收集缺失信息。 预订类型: {task_type} 所需信息: {required_slots} 当前槽位状态: {slots} 对话历史: {messages} 请遵循以下规则: {task_rules} 以自然、对话的方式回复。"""4.3 状态管理
from typing import TypedDict from langgraph.graph import StateGraph, START, END from langchain_core.messages import HumanMessage, AIMessage from datetime import datetime class SupervisorState(TypedDict): """用于管理整个系统状态的监督者状态""" messages: list[HumanMessage | AIMessage] task_type: str | None confidence: float slots: dict current_slot: str | None completed: bool stage: str4.4 图构建
4.4.1 主要节点定义
# 任务分类节点 def classify_task(state: SupervisorState) -> SupervisorState: """执行任务分类。""" print("\n=== 任务分类 ===") llm = ChatOpenAI(temperature=0) chain = PromptTemplate.from_template(TASK_CLASSIFICATION_TEMPLATE) | llm message = state["messages"][-1].content result = chain.invoke({"user_message": message}) classification = json.loads(result.content) state["task_type"] = classification["task"] state["confidence"] = classification["confidence"] state["stage"] = ( "slot_filling" if classification["confidence"] >= 0.5 else "classify" ) print(f"分类任务: {state['task_type']}") print(f"置信度: {state['confidence']:.2f}") return state # 初始化槽位节点 def initialize_slots(state: SupervisorState) -> SupervisorState: """根据任务类型初始化槽位。""" print("\n=== 初始化槽位 ===") if state["task_type"]: state["slots"] = { slot: "null" for slot in TASK_SLOTS[state["task_type"]].keys() } print(f"已初始化槽位: {state['slots']}") return state # 提取槽位节点 def extract_slots(state: SupervisorState) -> SupervisorState: """从对话中提取槽位信息。""" print("\n=== 提取槽位信息 ===") try: llm = ChatOpenAI(temperature=0) required_slots = "\n".join( [f"- {k}: {v}" for k, v in TASK_SLOTS[state["task_type"]].items()] ) messages_text = "\n".join(msg.content for msg in state["messages"]) last_message = state["messages"][-1].content current_date = datetime.now().strftime("%Y/%m/%d") chain = slot_extraction_prompt | llm result = chain.invoke( { "task_type": state["task_type"], "required_slots": required_slots, "slots": json.dumps(state["slots"], ensure_ascii=False), "messages": messages_text, "last_message": last_message, "current_date": current_date, } ) try: new_slots = json.loads(result.content) for slot, value in new_slots.items(): if ( value is not None and str(value).lower() != "null" and str(value).strip() ): state["slots"][slot] = value print("\n=== 当前槽位状态 ===") print(f"任务类型: {state['task_type']}") for slot, value in state["slots"].items(): print(f"{TASK_SLOTS[state['task_type']][slot]}: {value}") print("=====================\n") except json.JSONDecodeError: print("解析槽位信息时出错。") except Exception as e: print(f"提取槽位信息时出错: {str(e)}") return state # 生成响应节点 def generate_response(state: SupervisorState) -> SupervisorState: """根据当前状态生成响应。""" print("\n=== 生成响应 ===") response = "" if state["stage"] == "classify" and state["confidence"] < 0.5: response = "抱歉，我无法确定您正在寻找哪种类型的预订。\n" response += "可以进行以下预订:\n" for task in TASK_SLOTS.keys(): response += f"- {task}\n" response += ( "\n请更详细地说明您正在寻找的预订。" ) else: empty_slots = [] for slot, value in state["slots"].items(): if value is None or str(value).lower() == "null" or not str(value).strip(): empty_slots.append(slot) if empty_slots: task_type = state["task_type"] response = "请提供以下信息:\n" for slot in empty_slots: response += f"- {TASK_SLOTS[task_type][slot]}\n" else: response = ( "所有信息已输入。我将完成预订。" ) state["completed"] = True state["messages"].append(AIMessage(content=response)) print(f"\nAI: {response}") return state # 判断是否继续节点 def should_continue(state: SupervisorState) -> str: """确定下一步。""" print(f"\n=== 当前阶段: {state['stage']} ===") last_message = state["messages"][-1] if isinstance(last_message, AIMessage): print("等待用户输入。") return "generate_response" if state["stage"] == "classify": if state["confidence"] < 0.5: print("任务分类置信度低。重试。") return "generate_response" print("任务分类完成。继续进行槽位初始化。") return "initialize_slots" if not state["slots"]: print("槽位未初始化。初始化中。") return "initialize_slots" all_slots_filled = all( value is not None and str(value) != "null" and str(value).strip() != "" and str(value) != "undefined" for value in state["slots"].values() ) if all_slots_filled: print("所有槽位已填充。完成预订。") state["completed"] = True return "generate_response" print("需要更多信息。继续槽位提取。") return "extract_slots"4.4.2 创建预订代理图
def create_reservation_agent(): """创建集成的预订系统代理图。""" workflow = StateGraph(SupervisorState) workflow.add_node("classify", classify_task) workflow.add_node("initialize_slots", initialize_slots) workflow.add_node("extract_slots", extract_slots) workflow.add_node("generate_response", generate_response) workflow.add_edge(START, "classify") workflow.add_conditional_edges( "classify", should_continue, { "generate_response": "generate_response", "initialize_slots": "initialize_slots", }, ) workflow.add_edge("initialize_slots", "extract_slots") workflow.add_conditional_edges( "extract_slots", should_continue, {"generate_response": "generate_response", "extract_slots": "extract_slots"}, ) workflow.add_conditional_edges( "generate_response", should_continue, {"extract_slots": "extract_slots", "generate_response": END}, ) return workflow.compile() reservation_agent = create_reservation_agent()4.5 示例执行
def agent_chat(): """运行代理聊天交互""" print("\n=== AI 预订助手 ===") print("请随时讨论您想要的预订或请求。") print("\n📝 示例短语:") print("├── 餐厅: 我想为4个人预订下周五晚上7点的晚餐") print("├── 酒店: 我想在下个月1号到3号预订曼哈顿纽约的套房") print("├── 会议室: 我计划明天下午2点在市中心会议室开一个小时的会议") print("└── 航班: 我想预订下个月15号上午10点从洛杉矶到纽约的2张经济舱机票") print("\n要退出，请输入'quit'或'exit'。\n") messages_history = [] task_type = None slots = {} while True: try: user_input = input("User: ") if user_input.lower() in ["quit", "exit"]: print("退出预订系统。") break if not user_input.strip(): continue messages_history.append(HumanMessage(content=user_input)) state = { "messages": messages_history, "task_type": task_type, "confidence": 0.0, "slots": slots, "current_slot": None, "completed": False, "stage": "classify" } # 执行任务分类 state = reservation_agent.nodes["classify"].invoke(state) if state["confidence"] >= 0.5: task_type = state["task_type"] # 初始化槽位 if not slots: state = reservation_agent.nodes["initialize_slots"].invoke(state) slots = state["slots"] # 提取槽位信息 state = reservation_agent.nodes["extract_slots"].invoke(state) slots = state["slots"] # 生成响应 state = reservation_agent.nodes["generate_response"].invoke(state) if isinstance(state["messages"][-1], AIMessage): messages_history.append(state["messages"][-1]) all_slots_filled = all( value is not None and str(value) != "null" and str(value).strip() for value in slots.values() ) if all_slots_filled: print("\n=== 📝 对话摘要 ===") for msg in messages_history: prefix = "User: " if isinstance(msg, HumanMessage) else "AI: " print(f"{prefix}{msg.content}") print("\n=== ✨ 预订完成! ✨ ===") print(f"预订类型: {task_type}") for slot, value in slots.items(): print(f"{TASK_SLOTS[task_type][slot]}: {value}") print("\n要开始新的预订，请随时讨论。") print("\n要退出，请输入'quit'或'exit'。\n") messages_history = [] task_type = None slots = {} except Exception as e: print(f"发生错误: {str(e)}") print("请重试。") continue5. 案例效果
5.1 多任务支持
系统能够识别并处理多种类型的预订任务，包括餐厅预订、会议安排、酒店预订和航班预订。
5.2 动态槽位填充
系统能够根据用户输入自动提取相关信息，并识别缺失的必要信息。
5.3 自然对话交互
系统能够以自然、友好的方式与用户交互，收集缺失信息，提供流畅的用户体验。
5.4 信息验证
系统能够验证输入信息的有效性，如日期格式、数字格式等。
5.5 对话流程控制
系统能够根据当前状态智能控制对话流程，确保收集所有必要信息。
6. 案例实现思路
1任务分类设计
首先设计任务分类模块，通过分析用户输入识别预订类型，并评估分类置信度。
2槽位模型定义
为每种任务类型定义所需的槽位信息，包括名称、描述和验证规则。
3状态管理设计
设计状态管理机制，跟踪对话历史、任务类型、槽位状态和完成情况。
4信息提取逻辑
实现基于LLM的信息提取逻辑，从自然语言中提取结构化信息，并处理日期、数字等特殊格式。
5响应生成机制
设计响应生成机制，根据当前状态生成适当的回复，请求缺失信息或确认完成。
6工作流构建
使用LangGraph构建工作流图，定义节点、边和条件路由，实现对话流程控制。
7. 扩展建议
- 增加更多任务类型，如电影票预订、租车预订等
- 实现更复杂的信息验证逻辑，如日期合理性检查、座位可用性验证等
- 添加上下文记忆功能，支持多轮对话中的信息保持
- 集成外部API，实现真正的预订功能
- 添加多语言支持，扩展系统适用范围
- 实现更智能的对话策略，如主动推荐、替代方案提供等
- 添加用户偏好学习功能，提供个性化服务
- 实现对话可视化界面，提升用户体验
8. 总结
本案例展示了如何使用LangChain和LangGraph构建一个基于智能体的动态槽位填充系统。通过任务分类、槽位提取和响应生成等核心功能，系统能够以自然对话的方式收集必要信息，完成各种预订任务。
该案例的核心价值在于：
- 展示了动态槽位填充的实现方法
- 演示了多任务对话系统的构建
- 提供了自然语言信息提取的技术方案
- 展示了状态管理和流程控制的应用
这个案例为构建类似的智能对话系统提供了参考，特别是在需要收集结构化信息的场景中。通过扩展和优化，该系统可以应用于更广泛的领域，如客服机器人、智能助手、信息收集系统等。
