LangChain Intelligenter Kundenservice-Bot
Dieser Artikel integriert das zuvor Gelernte, um einen vollständigen intelligenten Kundenservice-Bot zu erstellen. Er kann die Wissensdatenbank abfragen, Bestellungen bearbeiten und bei Bedarf an einen menschlichen Mitarbeiter übergeben.
Anforderungsanalyse
| Merkmal | Implementierung |
|---|---|
| Fragen und Antworten zur Wissensdatenbank | RAG-Abruf + Modellantwort |
| Bestellabfrage | @tool Tool-Funktion |
| Konversationsgedächtnis | SqliteSaver Checkpointer |
| Filterung sensibler Inhalte | @before_model Middleware |
| Übergabe an einen menschlichen Mitarbeiter | HITL interrupt() / Command(resume=...) |
Umgebungs-Setup
Bevor Sie den eigentlichen Code schreiben, bereiten Sie zuerst die Laufzeitumgebung vor. Befolgen Sie die folgenden Schritte, dann ist sie in wenigen Minuten eingerichtet.
Schritt 1: Python-Version prüfen
Python 3.10 oder höher wird empfohlen. Öffnen Sie ein Terminal und geben Sie ein:
python --version
Wenn Ihre Version niedriger als 3.10 ist, wird empfohlen, Python zuerst zu aktualisieren, bevor Sie mit den nächsten Schritten fortfahren.
Schritt 2: Virtuelle Umgebung erstellen und aktivieren (empfohlen)
Um die Python-Umgebung des Systems nicht zu verunreinigen, wird empfohlen, für dieses Projekt eine separate virtuelle Umgebung zu erstellen:
# 创建虚拟环境 python -m venv venv # 激活虚拟环境（Windows） venv\Scripts\activate # 激活虚拟环境（macOS / Linux） source venv/bin/activate
Nach erfolgreicher Aktivierung zeigt die Terminal-Eingabeaufforderung Folgendes an(venv)Text.
Schritt 3: Abhängigkeiten installieren
Dieses Projekt verwendet insgesamt 6 Bibliotheken von Drittanbietern, die jeweils unterschiedliche Funktionen übernehmen. Es wird empfohlen, sie in der folgenden Reihenfolge einzeln zu installieren, damit sich bei einem Problem leichter feststellen lässt, welche Bibliothek fehlerhaft installiert wurde:
| Installationsbefehl | Zweck |
|---|---|
pip install langchain | LangChain-Kernbibliothek, die grundlegende Funktionen wie create_agent und @tool bereitstellt |
pip install langchain-deepseek | Ermöglicht init_chat_model, das Präfix "deepseek:" zu erkennen und das DeepSeek-Modell aufzurufen |
pip install langchain-openai | Stellt OpenAIEmbeddings zum Umwandeln von Wissensdatenbank-Text in Vektoren bereit |
pip install langchain-chroma chromadb | Lokale Vektordatenbank zum Speichern und Abrufen von Wissensdatenbank-Vektoren |
pip install langgraph-checkpoint-sqlite | SqliteSaver, persistiert den Konversationsspeicher in einer SQLite-Datei |
pip install python-dotenv | Liest Umgebungsvariablen wie API-Schlüssel aus der .env-Datei |
Du kannst sie auch alle mit einem einzigen Befehl installieren:
pip install langchain langchain-deepseek langchain-openai langchain-chroma chromadb langgraph-checkpoint-sqlite python-dotenv
Schritt 4: API-Schlüssel konfigurieren
Erstellen Sie eine neue Datei namens.envim Projektstammverzeichnis (beachte, dass der Dateiname mit einem Punkt beginnt und keine Erweiterung hat), und fülle den folgenden Inhalt ein:
# DeepSeek 官网申请：https://platform.deepseek.com DEEPSEEK_API_KEY=sk-你的deepseek密钥 # OpenAI 官网申请：https://platform.openai.com # 这里只用来调用 embedding 接口，不涉及 Chat 模型 OPENAI_API_KEY=sk-你的openai密钥
Wenn du keinen OpenAI-Schlüssel hast, kannst du stattdessen den von Alibaba Bailian verwenden..envDer Code für die .env-Datei lautet wie folgt:
# DeepSeek 官网申请：https://platform.deepseek.com DEEPSEEK_API_KEY=sk-你的deepseek密钥 # 阿里云百炼控制台申请：https://bailian.console.aliyun.com # 这里用来调用通义千问的 Embedding 服务，给知识库文本做向量化 DASHSCOPE_API_KEY=sk-你的百炼密钥
Die .env-Datei speichert private Schlüssel. Achte darauf, sie nicht in ein Git-Repository zu committen oder mit anderen zu teilen. Du kannst eine neue
.gitignoreDatei im Projekt anlegen und eine Zeile hinzufügen.envum versehentliche Commits zu vermeiden.
Schritt 5: Installation überprüfen
Erstellen Sie eine neuecheck_install.pyDatei, und führen Sie dann das untenstehende Skript aus, um zu überprüfen, ob die Abhängigkeiten und Schlüssel korrekt konfiguriert sind:
Beispiel
# Dateipfad: check_install.py
import os
from dotenv import load_dotenv
load_dotenv()
# Prüfen Sie, ob die Abhängigkeitspakete korrekt importiert werden können
import langchain
import langchain_deepseek
import langchain_openai
import langchain_chroma
import chromadb
import langgraph
print(f"langchain-Version: {langchain.__version__}")
# Prüfen Sie, ob die Schlüssel konfiguriert sind
assert os.getenv("DEEPSEEK_API_KEY"), "DEEPSEEK_API_KEY nicht erkannt, bitte überprüfen Sie die .env-Datei"
# assert os.getenv("OPENAI_API_KEY"), "OPENAI_API_KEY nicht erkannt, bitte überprüfen Sie die .env-Datei"
assert os.getenv("DASHSCOPE_API_KEY"), "DASHSCOPE_API_KEY nicht erkannt, bitte überprüfen Sie die .env-Datei"
print("Umgebungskonfiguration erfolgreich~ Jetzt können Sie mit dem Schreiben des Kundenservice-Bots beginnen!")
import os
from dotenv import load_dotenv
load_dotenv()
# Prüfen Sie, ob die Abhängigkeitspakete korrekt importiert werden können
import langchain
import langchain_deepseek
import langchain_openai
import langchain_chroma
import chromadb
import langgraph
print(f"langchain-Version: {langchain.__version__}")
# Prüfen Sie, ob die Schlüssel konfiguriert sind
assert os.getenv("DEEPSEEK_API_KEY"), "DEEPSEEK_API_KEY nicht erkannt, bitte überprüfen Sie die .env-Datei"
# assert os.getenv("OPENAI_API_KEY"), "OPENAI_API_KEY nicht erkannt, bitte überprüfen Sie die .env-Datei"
assert os.getenv("DASHSCOPE_API_KEY"), "DASHSCOPE_API_KEY nicht erkannt, bitte überprüfen Sie die .env-Datei"
print("Umgebungskonfiguration erfolgreich~ Jetzt können Sie mit dem Schreiben des Kundenservice-Bots beginnen!")
Ausführen:
python check_install.py
Wenn die Schlüssel korrekt konfiguriert sind, sieht die Ausgabe wie folgt aus:
langchain 版本: 1.3.0 环境配置成功~可以开始写客服机器人了！
Wenn ein Fehler auftritt, liegt das normalerweise daran, dass ein Paket nicht installiert ist oder die Schlüssel in der .env-Datei falsch sind. Überprüfen Sie den entsprechenden Schritt gemäß der Fehlermeldung.
Vollständiger Code
Nachdem die Umgebung eingerichtet ist (Abhängigkeitsinstallation und .env-Konfiguration werden im vorherigen Abschnitt "Einrichtung der Umgebung" behandelt), können Sie den vollständigen Code für den Kundenservice-Bot schreiben:
Beispiel
# Dateipfad: customer_service_bot.py
# Abhängigkeitsinstallation und .env-Konfiguration werden im vorherigen Abschnitt "Einrichtung der Umgebung" behandelt
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. Wissensdatenbank vorbereiten ==========
knowledge_base = [
"EXAMPLE, gegründet 2013, ist eine führende kostenlose Plattform zum Erlernen des Programmierens in China.",
"Die Plattform bietet über 300 Tutorials, die Python, Java, HTML, CSS, JavaScript und mehr abdecken.",
"Das Python3-Grundlagen-Tutorial hat 30 Kapitel, mit über 5 Millionen kumulierten Lernenden. Der Kurs ist völlig kostenlos.",
"Die VIP-Mitgliedschaft kostet ¥99/Monat oder ¥799/Jahr, inklusive Videokurse und Eins-zu-eins-Q&A-Support.",
"Rückerstattungsrichtlinie: Eine vollständige Rückerstattung ist innerhalb von 7 Tagen nach dem Kauf und innerhalb von 3 Lektionen möglich.",
"Die Plattform unterstützt eine Online-Programmierumgebung, mit der Sie Code schreiben und ausführen können, ohne Software installieren zu müssen.",
"Kundendienstzeiten: Montag bis Freitag 9:00-18:00, am Wochenende 10:00-16:00.",
]
# Verwenden Sie den Tongyi Qianwen Embedding-Dienst von Alibaba Cloud Bailian (DashScope)
# Bailians Embedding-API ist mit der OpenAI-API-Spezifikation kompatibel, daher verwenden wir direkt langchain-openai
# OpenAIEmbeddings und richten Sie base_url auf Bailians kompatiblen Endpunkt aus,
# es ist nicht nötig, langchain-community / dashscope zu installieren (dieses Paket wird nicht mehr gepflegt).
# text-embedding-v4 ist derzeit das empfohlene universelle Vektormodell und gibt standardmäßig 1024-dimensionale Vektoren aus.
#
# Zwei Schlüsselparameter sind erforderlich:
# - check_embedding_ctx_length=False: OpenAIEmbeddings verwendet standardmäßig tiktoken
# um Text vor dem Senden in ein Token-ID-Array vorzukodieren (die offizielle OpenAI-API akzeptiert dieses Format),
# aber Bailians kompatible API akzeptiert nur reine Zeichenketten. Wenn Sie diese Option nicht deaktivieren, wird
# ein Fehler "contents is neither str nor list of str" ausgelöst.
# - chunk_size=10: Die Embedding-API von Bailian akzeptiert höchstens 10 Textelemente pro Anfrage,
# während OpenAIEmbeddings standardmäßig 1000 Elemente auf einmal bündelt. Wenn die Wissensdatenbank etwas groß ist, wird das Limit überschritten und ein Fehler gemeldet.
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. Werkzeuge definieren ==========
@tool
def search_kb(query: str) -> str:
"""Durchsuche die EXAMPLE-Wissensdatenbank nach offiziellen Informationen über die Plattform, Kurse, Richtlinien usw.
Argumente:
query: Suchfrage oder Schlüsselwort
"""
docs = retriever.invoke(query)
if not docs:
return "Keine relevanten Informationen gefunden. Es wird empfohlen, an einen menschlichen Mitarbeiter weiterzuleiten."
return "\n".join(f"- {doc.page_content}" for doc in docs)
# Mock-Bestellungsdatenbank
orders_db = {
"ORD-2024-001": {"user": "Xiaoming", "item": "VIP-Jahresmitgliedschaft",
"amount": 799, "status": "Abgeschlossen", "date": "2024-01-15"},
"ORD-2024-002": {"user": "Xiaoming", "item": "Python-Praxiskurs",
"amount": 199, "status": "Unterwegs", "date": "2024-03-20"},
}
@tool
def query_order(order_id: str) -> str:
"""Fragt den Bestellstatus und die Details anhand der Bestellnummer ab.
Argumente:
order_id: Bestellnummer, z. B. ORD-2024-001
"""
order = orders_db.get(order_id.upper())
if not order:
return f"Bestellung {order_id} nicht gefunden. Bitte überprüfen Sie die Bestellnummer."
return (f"Bestellung {order_id}: {order['item']} | "
f"Betrag ¥{order['amount']} | "
f"Status {order['status']} | "
f"Datum {order['date']}")
@tool
def transfer_to_human(reason: str) -> str:
"""Übergeben Sie den Benutzer an einen menschlichen Mitarbeiter.
Argumente:
reason: Grund für die Übergabe
"""
approval = interrupt({
"action": "transfer_to_human",
"reason": reason,
"message": f"Der Benutzer bittet darum, an einen menschlichen Mitarbeiter weitergeleitet zu werden. Grund: {reason}. Möchten Sie die Weiterleitung vornehmen?"
})
if approval.get("confirmed"):
return (f"Sie wurden an einen menschlichen Mitarbeiter weitergeleitet. Geschätzte Wartezeit: {approval.get('wait_time', 3)} Minuten."
f"Ticketnummer: TK-{approval.get('ticket_id', 'N/A')}")
return "Die Weiterleitung wurde storniert, ich werde Sie weiterhin betreuen."
# ========== 3. Middleware definieren ==========
@before_model
def content_guard(state, runtime):
"""Unangemessene Inhalte in Benutzereingaben filtern"""
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="Entschuldigung, ich kann diese Anfrage nicht verarbeiten.")]
}
return None
@after_model
def auto_signature(state, runtime):
"""Kundenservice-Signatur automatisch anhängen"""
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
):
# Wichtig: last.id wiederverwenden, damit der add_messages-Reducer die Nachricht an Ort und Stelle ersetzt,
# statt sie als neue Nachricht an die Historie anzuhängen (andernfalls wächst die Historie immer weiter,
# wobei pro Runde eine "unsignierte Version" und eine "signierte Version" hinzugefügt werden)
return {"messages": [AIMessage(
id=last.id,
content=last.content
+ "\n\n---\n"EXAMPLE Kundenservice-Center | Öffnungszeiten: 9:00-18:00"
)]}
return None
# ========== 4. Agent erstellen ==========
# SqliteSaver.from_conn_string() gibt einen Context-Manager zurück, der nur für "Schließen nach Gebrauch" geeignet ist
# Einmal-Skripte. Ein Kundenservice-Bot muss dieselbe Datenbankverbindung über mehrere chat()-Aufrufe hinweg beibehalten,
# daher stellen wir hier die Verbindung selbst her und übergeben sie an den SqliteSaver-Konstruktor.
# check_same_thread=False liegt daran, dass Web-Frameworks normalerweise dieselbe Verbindung über Threads hinweg verwenden.
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="""Du bist "Xiao Cai", der intelligente Kundenservice-Agent von EXAMPLE.
## Deine Aufgaben
Empfange jeden Nutzer herzlich und sprich ihn mit "Sie" an (die höfliche Form von "you")
2. Verwende für Fragen zu Plattforminformationen, Kursinhalten, Richtlinien usw. search_kb, um sie nachzuschlagen
3. Verwende für Bestellanfragen das Tool query_order
4. Wenn du auf Probleme stößt, die du nicht lösen kannst, verwende transfer_to_human, um an einen menschlichen Mitarbeiter weiterzuleiten
## Verhaltenskodex
- Halte die Antworten kurz, jeweils 2-3 Sätze
- Wenn du etwas nicht weißt, frage die Wissensdatenbank ab; wenn nichts gefunden wird, sag es dem Nutzer ehrlich
- Pflege einen freundlichen und warmen Ton""",
)
# ========== 5. Konversationsschnittstelle ==========
def chat(thread_id: str, message: str) -> str:
"""Verarbeitet Benutzernachrichten und gibt Antworten zurück"""
config = {"configurable": {"thread_id": thread_id}}
# Agent ausführen
result = agent.invoke(
{"messages": [HumanMessage(content=message)]},
config=config,
)
# Prüfe, ob eine Übergabe erforderlich ist (HITL)
state = .get_state(config)
if state.tasks and state.tasks[0].interrupts:
interrupt_info = state.tasks[0].interrupts[0].value
return f"[Genehmigung erforderlich] {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:
"""Nachdem der menschliche Agent im Backend zugestimmt hat, den durch interrupt() unterbrochenen Transfervorgang fortsetzen.
Dies entspricht den Genehmigungsdaten, die im Tool transfer_to_human warten.
"""
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: Wissensdatenbank-Abfrage ===")
print(chat(user_id, "Wie viele Kapitel hat das Python3-Tutorial?"))
print()
print("=== Test 2: Bestellabfrage ===")
print(chat(user_id, "Wie lautet der Status meiner Bestellung ORD-2024-001?"))
print()
print("=== Test 3: VIP-Beratung ===")
print(chat(user_id, "Wie viel kostet eine VIP-Mitgliedschaft?"))
print()
print("=== Test 4: Gedächtnis testen ===")
print(chat(user_id, "Welche Frage habe ich gerade gestellt?"))
print()
print("=== Test 5: Menschliche Übergabe (HITL) ===")
print(chat(user_id, "Ich möchte eine Beschwerde einreichen, bitte leiten Sie mich an einen menschlichen Mitarbeiter weiter.")) # Unterbrechung auslösen, auf Genehmigung warten
print(resume_transfer(user_id, confirmed=True,
wait_time=5, ticket_id="8823")) # Nach Backend-Bestätigung fortsetzen
conn.close()
# Abhängigkeitsinstallation und .env-Konfiguration werden im vorherigen Abschnitt "Einrichtung der Umgebung" behandelt
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. Wissensdatenbank vorbereiten ==========
knowledge_base = [
"EXAMPLE, gegründet 2013, ist eine führende kostenlose Plattform zum Erlernen des Programmierens in China.",
"Die Plattform bietet über 300 Tutorials, die Python, Java, HTML, CSS, JavaScript und mehr abdecken.",
"Das Python3-Grundlagen-Tutorial hat 30 Kapitel, mit über 5 Millionen kumulierten Lernenden. Der Kurs ist völlig kostenlos.",
"Die VIP-Mitgliedschaft kostet ¥99/Monat oder ¥799/Jahr, inklusive Videokurse und Eins-zu-eins-Q&A-Support.",
"Rückerstattungsrichtlinie: Eine vollständige Rückerstattung ist innerhalb von 7 Tagen nach dem Kauf und innerhalb von 3 Lektionen möglich.",
"Die Plattform unterstützt eine Online-Programmierumgebung, mit der Sie Code schreiben und ausführen können, ohne Software installieren zu müssen.",
"Kundendienstzeiten: Montag bis Freitag 9:00-18:00, am Wochenende 10:00-16:00.",
]
# Verwenden Sie den Tongyi Qianwen Embedding-Dienst von Alibaba Cloud Bailian (DashScope)
# Bailians Embedding-API ist mit der OpenAI-API-Spezifikation kompatibel, daher verwenden wir direkt langchain-openai
# OpenAIEmbeddings und richten Sie base_url auf Bailians kompatiblen Endpunkt aus,
# es ist nicht nötig, langchain-community / dashscope zu installieren (dieses Paket wird nicht mehr gepflegt).
# text-embedding-v4 ist derzeit das empfohlene universelle Vektormodell und gibt standardmäßig 1024-dimensionale Vektoren aus.
#
# Zwei Schlüsselparameter sind erforderlich:
# - check_embedding_ctx_length=False: OpenAIEmbeddings verwendet standardmäßig tiktoken
# um Text vor dem Senden in ein Token-ID-Array vorzukodieren (die offizielle OpenAI-API akzeptiert dieses Format),
# aber Bailians kompatible API akzeptiert nur reine Zeichenketten. Wenn Sie diese Option nicht deaktivieren, wird
# ein Fehler "contents is neither str nor list of str" ausgelöst.
# - chunk_size=10: Die Embedding-API von Bailian akzeptiert höchstens 10 Textelemente pro Anfrage,
# während OpenAIEmbeddings standardmäßig 1000 Elemente auf einmal bündelt. Wenn die Wissensdatenbank etwas groß ist, wird das Limit überschritten und ein Fehler gemeldet.
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. Werkzeuge definieren ==========
@tool
def search_kb(query: str) -> str:
"""Durchsuche die EXAMPLE-Wissensdatenbank nach offiziellen Informationen über die Plattform, Kurse, Richtlinien usw.
Argumente:
query: Suchfrage oder Schlüsselwort
"""
docs = retriever.invoke(query)
if not docs:
return "Keine relevanten Informationen gefunden. Es wird empfohlen, an einen menschlichen Mitarbeiter weiterzuleiten."
return "\n".join(f"- {doc.page_content}" for doc in docs)
# Mock-Bestellungsdatenbank
orders_db = {
"ORD-2024-001": {"user": "Xiaoming", "item": "VIP-Jahresmitgliedschaft",
"amount": 799, "status": "Abgeschlossen", "date": "2024-01-15"},
"ORD-2024-002": {"user": "Xiaoming", "item": "Python-Praxiskurs",
"amount": 199, "status": "Unterwegs", "date": "2024-03-20"},
}
@tool
def query_order(order_id: str) -> str:
"""Fragt den Bestellstatus und die Details anhand der Bestellnummer ab.
Argumente:
order_id: Bestellnummer, z. B. ORD-2024-001
"""
order = orders_db.get(order_id.upper())
if not order:
return f"Bestellung {order_id} nicht gefunden. Bitte überprüfen Sie die Bestellnummer."
return (f"Bestellung {order_id}: {order['item']} | "
f"Betrag ¥{order['amount']} | "
f"Status {order['status']} | "
f"Datum {order['date']}")
@tool
def transfer_to_human(reason: str) -> str:
"""Übergeben Sie den Benutzer an einen menschlichen Mitarbeiter.
Argumente:
reason: Grund für die Übergabe
"""
approval = interrupt({
"action": "transfer_to_human",
"reason": reason,
"message": f"Der Benutzer bittet darum, an einen menschlichen Mitarbeiter weitergeleitet zu werden. Grund: {reason}. Möchten Sie die Weiterleitung vornehmen?"
})
if approval.get("confirmed"):
return (f"Sie wurden an einen menschlichen Mitarbeiter weitergeleitet. Geschätzte Wartezeit: {approval.get('wait_time', 3)} Minuten."
f"Ticketnummer: TK-{approval.get('ticket_id', 'N/A')}")
return "Die Weiterleitung wurde storniert, ich werde Sie weiterhin betreuen."
# ========== 3. Middleware definieren ==========
@before_model
def content_guard(state, runtime):
"""Unangemessene Inhalte in Benutzereingaben filtern"""
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="Entschuldigung, ich kann diese Anfrage nicht verarbeiten.")]
}
return None
@after_model
def auto_signature(state, runtime):
"""Kundenservice-Signatur automatisch anhängen"""
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
):
# Wichtig: last.id wiederverwenden, damit der add_messages-Reducer die Nachricht an Ort und Stelle ersetzt,
# statt sie als neue Nachricht an die Historie anzuhängen (andernfalls wächst die Historie immer weiter,
# wobei pro Runde eine "unsignierte Version" und eine "signierte Version" hinzugefügt werden)
return {"messages": [AIMessage(
id=last.id,
content=last.content
+ "\n\n---\n"EXAMPLE Kundenservice-Center | Öffnungszeiten: 9:00-18:00"
)]}
return None
# ========== 4. Agent erstellen ==========
# SqliteSaver.from_conn_string() gibt einen Context-Manager zurück, der nur für "Schließen nach Gebrauch" geeignet ist
# Einmal-Skripte. Ein Kundenservice-Bot muss dieselbe Datenbankverbindung über mehrere chat()-Aufrufe hinweg beibehalten,
# daher stellen wir hier die Verbindung selbst her und übergeben sie an den SqliteSaver-Konstruktor.
# check_same_thread=False liegt daran, dass Web-Frameworks normalerweise dieselbe Verbindung über Threads hinweg verwenden.
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="""Du bist "Xiao Cai", der intelligente Kundenservice-Agent von EXAMPLE.
## Deine Aufgaben
Empfange jeden Nutzer herzlich und sprich ihn mit "Sie" an (die höfliche Form von "you")
2. Verwende für Fragen zu Plattforminformationen, Kursinhalten, Richtlinien usw. search_kb, um sie nachzuschlagen
3. Verwende für Bestellanfragen das Tool query_order
4. Wenn du auf Probleme stößt, die du nicht lösen kannst, verwende transfer_to_human, um an einen menschlichen Mitarbeiter weiterzuleiten
## Verhaltenskodex
- Halte die Antworten kurz, jeweils 2-3 Sätze
- Wenn du etwas nicht weißt, frage die Wissensdatenbank ab; wenn nichts gefunden wird, sag es dem Nutzer ehrlich
- Pflege einen freundlichen und warmen Ton""",
)
# ========== 5. Konversationsschnittstelle ==========
def chat(thread_id: str, message: str) -> str:
"""Verarbeitet Benutzernachrichten und gibt Antworten zurück"""
config = {"configurable": {"thread_id": thread_id}}
# Agent ausführen
result = agent.invoke(
{"messages": [HumanMessage(content=message)]},
config=config,
)
# Prüfe, ob eine Übergabe erforderlich ist (HITL)
state = .get_state(config)
if state.tasks and state.tasks[0].interrupts:
interrupt_info = state.tasks[0].interrupts[0].value
return f"[Genehmigung erforderlich] {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:
"""Nachdem der menschliche Agent im Backend zugestimmt hat, den durch interrupt() unterbrochenen Transfervorgang fortsetzen.
Dies entspricht den Genehmigungsdaten, die im Tool transfer_to_human warten.
"""
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: Wissensdatenbank-Abfrage ===")
print(chat(user_id, "Wie viele Kapitel hat das Python3-Tutorial?"))
print()
print("=== Test 2: Bestellabfrage ===")
print(chat(user_id, "Wie lautet der Status meiner Bestellung ORD-2024-001?"))
print()
print("=== Test 3: VIP-Beratung ===")
print(chat(user_id, "Wie viel kostet eine VIP-Mitgliedschaft?"))
print()
print("=== Test 4: Gedächtnis testen ===")
print(chat(user_id, "Welche Frage habe ich gerade gestellt?"))
print()
print("=== Test 5: Menschliche Übergabe (HITL) ===")
print(chat(user_id, "Ich möchte eine Beschwerde einreichen, bitte leiten Sie mich an einen menschlichen Mitarbeiter weiter.")) # Unterbrechung auslösen, auf Genehmigung warten
print(resume_transfer(user_id, confirmed=True,
wait_time=5, ticket_id="8823")) # Nach Backend-Bestätigung fortsetzen
conn.close()
Ausführungsergebnisse:
=== 测试 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/月 oder ¥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
Projektzusammenfassung
Dieser Kundenservice-Bot integriert die folgenden LangChain-Funktionen:
| Merkmal | Verwendung im Projekt |
|---|---|
| RAG-Retrieval | search_kb-Tool + Chroma-Vektorspeicher |
| Tool-Aufruf | query_order, transfer_to_human |
| Checkpointer | SqliteSaver persistiert Konversationen und ermöglicht so ein Multi-Turn-Gedächtnis |
| Middleware | before_model Inhaltsfilterung + after_model Signatur anhängen (message id wiederverwenden, um an Ort und Stelle zu ersetzen) |
| HITL | interrupt() pausiert die Ausführung + Command(resume=...) setzt nach Genehmigung fort, wodurch eine vollständige Schleife zur Übergabe an einen Menschen erreicht wird |