LangChain Intelligent Customer Service Bot

This article integrates the knowledge learned earlier to build a complete intelligent customer service bot. It can query the knowledge base, handle orders, and transfer to a human agent when necessary.

Requirements Analysis

Feature	Implementation
Knowledge Base Q&A	RAG retrieval + model response
Order Inquiry	@tool Tool Function
Conversation Memory	SqliteSaver Checkpointer
Sensitive Content Filtering	@before_model Middleware
Transfer to Human Agent	HITL interrupt() / Command(resume=...)

Environment Setup

Before writing the actual code, prepare the runtime environment first. Follow the steps below and you'll have it set up in a few minutes.

Step 1: Check Python Version

Python 3.10 or higher is recommended. Open a terminal and enter:

python --version

If your version is lower than 3.10, it is recommended to upgrade Python first before proceeding with the next steps.

Step 2: Create and Activate a Virtual Environment (Recommended)

To avoid polluting the system's Python environment, it is recommended to create a separate virtual environment for this project:

# 创建虚拟环境
python -m venv venv

# 激活虚拟环境（Windows）
venv\Scripts\activate

# 激活虚拟环境（macOS / Linux）
source venv/bin/activate

After successful activation, the terminal prompt will display the(venv)text.

Step 3: Install Dependencies

This project uses a total of 6 third-party libraries, each responsible for different functions. It is recommended to install them one by one in the following order, making it easier to identify which library is incorrectly installed when a problem occurs:

Installation Command	Purpose
`pip install langchain`	LangChain core library, providing basic capabilities such as create_agent and @tool
`pip install langchain-deepseek`	Enables init_chat_model to recognize the "deepseek:" prefix and call the DeepSeek model
`pip install langchain-openai`	Provides OpenAIEmbeddings for converting knowledge base text into vectors
`pip install langchain-chroma chromadb`	Local vector database for storing and retrieving knowledge base vectors
`pip install langgraph-checkpoint-sqlite`	SqliteSaver, persists conversation memory to a SQLite file
`pip install python-dotenv`	Reads environment variables such as API keys from the .env file

You can also install them all with a single command:

pip install langchain langchain-deepseek langchain-openai langchain-chroma chromadb langgraph-checkpoint-sqlite python-dotenv

Step 4: Configure API Keys

Create a new file named.envin the project root directory (note the filename starts with a dot and has no extension), and fill in the following content:

# DeepSeek 官网申请：https://platform.deepseek.com
DEEPSEEK_API_KEY=sk-你的deepseek密钥

# OpenAI 官网申请：https://platform.openai.com
# 这里只用来调用 embedding 接口，不涉及 Chat 模型
OPENAI_API_KEY=sk-你的openai密钥

If you don't have an OpenAI key, you can use Alibaba Bailian's instead..envThe code for the .env file is as follows:

# DeepSeek 官网申请：https://platform.deepseek.com
DEEPSEEK_API_KEY=sk-你的deepseek密钥

# 阿里云百炼控制台申请：https://bailian.console.aliyun.com
# 这里用来调用通义千问的 Embedding 服务，给知识库文本做向量化
DASHSCOPE_API_KEY=sk-你的百炼密钥

The .env file stores private keys. Be sure not to commit it to a Git repository or share it with others. You can create a new.gitignorefile in the project and add a line.envto avoid accidental commits.

Step 5: Verify Installation

Create a newcheck_install.pyfile, then run the script below to check whether the dependencies and keys are configured correctly:

Example

# File path: check_install.py
import os
from dotenv import load_dotenv

load_dotenv()

# Check that the dependency packages can be imported properly
import langchain
import langchain_deepseek
import langchain_openai
import langchain_chroma
import chromadb
import langgraph

print(f"langchain version: {langchain.__version__}")

# Check whether the keys are configured
assert os.getenv("DEEPSEEK_API_KEY"), "DEEPSEEK_API_KEY not detected, please check the .env file"
# assert os.getenv("OPENAI_API_KEY"), "OPENAI_API_KEY not detected, please check the .env file"
assert os.getenv("DASHSCOPE_API_KEY"), "DASHSCOPE_API_KEY not detected, please check the .env file"

print("Environment configuration successful~ Now you can start writing the customer service bot!")

Run:

python check_install.py

If the keys are configured correctly, the output is as follows:

langchain 版本: 1.3.0
环境配置成功~可以开始写客服机器人了！

If an error occurs, it is usually because a package is not installed or the keys in the .env file are incorrect. Check the corresponding step according to the error message.

Complete Code

After the environment is set up (dependency installation and .env configuration are covered in the previous section "Environment Setup"), you can write the complete customer service bot code:

Example

# File path: customer_service_bot.py
# Dependency installation and .env configuration are covered in the previous section "Environment Setup"
from dotenv import load_dotenv
load_dotenv()

import os
import sqlite3
from typing import Annotated
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.agents.middleware import before_model, after_model
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage, AIMessage
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.types import interrupt, Command

# ========== 1. Prepare the Knowledge Base ==========

knowledge_base = [
"EXAMPLE, founded in 2013, is a leading free programming learning platform in China.",
"The platform offers 300+ tutorials, covering Python, Java, HTML, CSS, JavaScript, and more.",
"The Python3 basic tutorial has 30 chapters, with over 5 million cumulative learners. The course is completely free.",
"VIP membership costs ¥99/month or ¥799/year, including video courses and one-on-one Q&A support.",
"Refund policy: A full refund is available within 7 days of purchase and within 3 lessons.",
"The platform supports an online programming environment, allowing you to write and run code without installing any software.",
"Customer service hours: Monday to Friday 9:00-18:00, weekends 10:00-16:00.",
]

# Use Alibaba Cloud Bailian (DashScope)'s Tongyi Qianwen Embedding service
# Bailian's Embedding API is compatible with the OpenAI API specification, so we directly use langchain-openai
# OpenAIEmbeddings and point base_url to Bailian's compatible endpoint,
# there is no need to install langchain-community / dashscope (this package is no longer maintained).
# text-embedding-v4 is currently the recommended general-purpose vector model, outputting 1024-dimensional vectors by default.
#
# Two key parameters are required:
# - check_embedding_ctx_length=False: OpenAIEmbeddings by default uses tiktoken
# to pre-encode text into a token id array before sending (the official OpenAI API accepts this format),
# but Bailian's compatible API only accepts raw strings. If you don't disable this option, it will throw
# a "contents is neither str nor list of str" error.
# - chunk_size=10: Bailian's Embedding API accepts at most 10 text items per request,
# while OpenAIEmbeddings bundles 1000 items at a time by default. If the knowledge base is slightly large, it will exceed the limit and report an error.
embeddings = OpenAIEmbeddings(
model="text-embedding-v4",
api_key=os.getenv("DASHSCOPE_API_KEY"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
check_embedding_ctx_length=False,
chunk_size=10,
)
chunks = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=30
).create_documents(knowledge_base)
vector_store = Chroma.from_documents(chunks, embeddings)
retriever = vector_store.as_retriever(search_kwargs={"k": 3})

# ========== 2. Define Tools ==========

@tool
def search_kb(query: str) -> str:
"""Search the EXAMPLE knowledge base for official information about the platform, courses, policies, etc.

Args:
query: search question or keyword
"""
docs = retriever.invoke(query)
if not docs:
return "No relevant information found. It is recommended to transfer to a human agent."
return "\n".join(f"- {doc.page_content}" for doc in docs)

# Mock order database
orders_db = {
"ORD-2024-001": {"user": "Xiaoming", "item": "VIP annual membership",
"amount": 799, "status": "Completed", "date": "2024-01-15"},
"ORD-2024-002": {"user": "Xiaoming", "item": "Python practical course",
"amount": 199, "status": "In transit", "date": "2024-03-20"},
}

@tool
def query_order(order_id: str) -> str:
"""Query the order status and details by order number.

Args:
order_id: order number, e.g., ORD-2024-001
"""
order = orders_db.get(order_id.upper())
if not order:
return f"Order {order_id} not found. Please verify the order number."
return (f"Order {order_id}: {order['item']} | "
f"Amount ¥{order['amount']} | "
f"Status {order['status']} | "
f"Date {order['date']}")

@tool
def transfer_to_human(reason: str) -> str:
"""Transfer the user to a human agent.

Args:
reason: reason for transfer
"""
approval = interrupt({
"action": "transfer_to_human",
"reason": reason,
"message": f"The user requests to be transferred to a human agent. Reason: {reason}. Do you want to transfer?"
})
if approval.get("confirmed"):
return (f"You have been transferred to a human agent. Estimated wait time: {approval.get('wait_time', 3)} minutes."
f"Ticket Number: TK-{approval.get('ticket_id', 'N/A')}")
return "The transfer has been cancelled, I will continue to serve you."

# ========== 3. Define Middleware ==========

@before_model
def content_guard(state, runtime):
"""Filter inappropriate content in user input"""
last_msg = state["messages"][-1] if state.get("messages") else None
if not last_msg:
return None
content = str(getattr(last_msg, 'content', ''))
blocked = ["Huang X", "X Bo", "illegal"]
for word in blocked:
if word in content:
return {
"jump_to": "end",
"messages": [HumanMessage(content="Sorry, I cannot process this request.")]
}
return None

@after_model
def auto_signature(state, runtime):
"""Automatically append customer service signature"""
msgs = state.get("messages", [])
if not msgs:
return None
last = msgs[-1]
if last.type == "ai" and last.content and not (
hasattr(last, 'tool_calls') and last.tool_calls
):
# Key: reuse last.id so the add_messages reducer replaces the message in place,
# rather than appending it as a new message to the history (otherwise the history will keep growing,
# each round adding one "unsigned version" and one "signed version")
return {"messages": [AIMessage(
id=last.id,
content=last.content
+ "\n\n---\n"EXAMPLE Customer Service Center | Working hours: 9:00-18:00"
)]}
return None

# ========== 4. Create Agent ==========

# SqliteSaver.from_conn_string() returns a context manager, suitable only for "close after use"
# one-off scripts. A customer service bot needs to keep the same database connection across multiple chat() calls,
# so here we establish the connection ourselves and pass it to the SqliteSaver constructor.
# check_same_thread=False is because web frameworks usually call the same connection across threads.
conn = sqlite3.connect("customer_service.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)

model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)

agent = create_agent(
model=model,
tools=[search_kb, query_order, transfer_to_human],
middleware=[content_guard, auto_signature],
checkpointer=checkpointer,
system_prompt="""You are "Xiao Cai", the intelligent customer service agent of EXAMPLE.

## Your Responsibilities
Warmly receive every user and address them with "you" (the polite form of "you")
2. For questions about platform information, course content, policies, etc., use search_kb to look them up
3. For order inquiries, use the query_order tool
4. When encountering problems you cannot solve, use transfer_to_human to transfer to a human agent

## Code of Conduct
- Keep answers concise, 2-3 sentences each time
- If you don't know, query the knowledge base; if not found, honestly tell the user
- Maintain a friendly and warm tone""",
)

# ========== 5. Conversation Interface ==========

def chat(thread_id: str, message: str) -> str:
"""Process user messages and return replies"""
config = {"configurable": {"thread_id": thread_id}}

# Run the Agent
result = agent.invoke(
{"messages": [HumanMessage(content=message)]},
config=config,
)

# Check whether transfer is needed (HITL)
state = agent.get_state(config)
if state.tasks and state.tasks[0].interrupts:
interrupt_info = state.tasks[0].interrupts[0].value
return f"[Approval Required] {interrupt_info.get('message', '')}"

return result["messages"][-1].content

def resume_transfer(thread_id: str, confirmed: bool,
wait_time: int = 3, ticket_id: str = "0001") -> str:
"""After the human agent approves in the backend, resume the transfer process interrupted by interrupt().

This corresponds to the approval data waiting in the transfer_to_human tool.
"""
config = {"configurable": {"thread_id": thread_id}}
result = agent.invoke(
Command(resume={
"confirmed": confirmed,
"wait_time": wait_time,
"ticket_id": ticket_id,
}),
config=config,
)
return result["messages"][-1].content

# ========== 6. Tests ==========

if __name__ == "__main__":
user_id = "user_xiaoming"

print("=== Test 1: Knowledge Base Query ===")
print(chat(user_id, "How many chapters does the Python3 tutorial have?"))
print()

print("=== Test 2: Order Query ===")
print(chat(user_id, "What is the status of my order ORD-2024-001?"))
print()

print("=== Test 3: VIP Consultation ===")
print(chat(user_id, "How much is a VIP membership?"))
print()

print("=== Test 4: Test Memory ===")
print(chat(user_id, "What question did I just ask?"))
print()

print("=== Test 5: Human Transfer (HITL) ===")
print(chat(user_id, "I want to file a complaint, please transfer me to a human agent.")) # Trigger interrupt, wait for approval
print(resume_transfer(user_id, confirmed=True,
wait_time=5, ticket_id="8823")) # Resume after backend confirmation

conn.close()

Run results:

=== 测试 1：知识库查询 ===
您好！Example的 **Python3 基础教程共 30 章**，完全免费，累计学习人次已超 500 万哦！如需了解其他教程，也可以随时问我～

---
Example 客服中心 | 工作时间 9:00-18:00

=== 测试 2：订单查询 ===
您好！您查询的订单 **ORD-2024-001** 状态为 **已完成**。订单内容是 **VIP 年费会员**，金额 **¥799**，下单日期为 **2024-01-15**。请问还有什么可以帮您的吗？

---
Example 客服中心 | 工作时间 9:00-18:00

=== 测试 3：VIP 咨询 ===
您好！VIP 会员有 **¥99/月** 和 **¥799/年** 两种套餐，包含视频课程和一对一答疑服务哦～请问您需要办理哪种呢？

---
Example 客服中心 | 工作时间 9:00-18:00

---
Example 客服中心 | 工作时间 9:00-18:00

=== 测试 4：测试记忆 ===
您好！您刚才问了以下三个问题：
1. **Python3 教程有多少章？** —— 共 30 章，完全免费
2. **我的订单 ORD-2024-001 状态是什么？** —— 状态为"已完成"
3. **VIP 会员多少钱？** —— ¥99/月 或 ¥799/年

还有什么需要我帮忙的吗？

---
Example 客服中心 | 工作时间 9:00-18:00

---
Example 客服中心 | 工作时间 9:00-18:00

=== 测试 5：人工转接（HITL） ===
[需要审批] 用户请求转接人工客服，原因：用户要求投诉，需要转接人工客服处理。是否转接？
您好！已经为您转接人工客服，预计等待约 5 分钟。您的工单号是 **TK-8823**，请稍候，客服会尽快为您处理～

---
Example 客服中心 | 工作时间 9:00-18:00

---
Example 客服中心 | 工作时间 9:00-18:00

Project Summary

This customer service bot integrates the following LangChain features:

Feature	Usage in the project
RAG Retrieval	search_kb tool + Chroma vector store
Tool calling	query_order、transfer_to_human
Checkpointer	SqliteSaver persists conversations, enabling multi-turn memory
Middleware	before_model content filtering + after_model signature appending (reuse message id to replace in place)
HITL	interrupt() pauses execution + Command(resume=...) resumes after approval, achieving a complete human transfer loop

Other extensions

Agent Development Configuration

Reference Manual

LangChain Intelligent Customer Service Bot

Requirements Analysis

Environment Setup

Step 1: Check Python Version

Step 2: Create and Activate a Virtual Environment (Recommended)

Step 3: Install Dependencies

Step 4: Configure API Keys

Step 5: Verify Installation

Example

Complete Code

Example

Project Summary