Back to Details

ai-building-chatbots

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

Build a Conversational AI Chatbot

构建对话式AI聊天机器人

Guide the user through building a multi-turn chatbot that remembers context, follows conversation flows, and produces high-quality responses. Uses DSPy for optimizable response generation and LangGraph for conversation state, memory, and flow control.
引导用户构建一款能够记住上下文、遵循对话流程并生成高质量响应的多轮对话聊天机器人。使用DSPy实现可优化的响应生成，使用LangGraph管理对话状态、记忆与流程控制。

Step 1: Define the conversation

步骤1：定义对话场景

Ask the user:
  1. What does the bot do? (answer questions, resolve issues, qualify leads, guide onboarding?)
  2. What states can the conversation be in? (greeting, gathering info, resolving, escalating, closing?)
  3. When should the bot escalate to a human? (complex issues, angry users, sensitive topics?)
  4. What docs or data should it draw from? (help articles, product docs, FAQs, database?)
询问用户：
  1. 机器人的用途是什么？（回答问题、解决问题、潜在客户资质审核、引导新用户入门？）
  2. 对话可能处于哪些状态？（问候、收集信息、解决问题、升级处理、结束对话？）
  3. 机器人何时需要将对话转交给人工？（复杂问题、用户情绪激动、敏感话题？）
  4. 机器人需要调用哪些文档或数据？（帮助文章、产品文档、FAQ、数据库？）

Step 2: Build the response module (DSPy)

步骤2：构建响应模块（DSPy）

The core of your chatbot is a DSPy module that generates responses given conversation history and context.
聊天机器人的核心是一个DSPy模块，可根据对话历史和上下文生成响应。

Basic response module

基础响应模块

python
import dspy

class ChatResponse(dspy.Signature):
    """Generate a helpful, on-brand response to the user's message."""
    conversation_history: str = dspy.InputField(desc="Previous messages in the conversation")
    context: str = dspy.InputField(desc="Relevant information from docs or database")
    user_message: str = dspy.InputField(desc="The user's latest message")
    response: str = dspy.OutputField(desc="Helpful response to the user")

class ChatBot(dspy.Module):
    def __init__(self):
        self.respond = dspy.ChainOfThought(ChatResponse)

    def forward(self, conversation_history, context, user_message):
        return self.respond(
            conversation_history=conversation_history,
            context=context,
            user_message=user_message,
        )
python
import dspy

class ChatResponse(dspy.Signature):
    """Generate a helpful, on-brand response to the user's message."""
    conversation_history: str = dspy.InputField(desc="Previous messages in the conversation")
    context: str = dspy.InputField(desc="Relevant information from docs or database")
    user_message: str = dspy.InputField(desc="The user's latest message")
    response: str = dspy.OutputField(desc="Helpful response to the user")

class ChatBot(dspy.Module):
    def __init__(self):
        self.respond = dspy.ChainOfThought(ChatResponse)

    def forward(self, conversation_history, context, user_message):
        return self.respond(
            conversation_history=conversation_history,
            context=context,
            user_message=user_message,
        )

With intent classification

带意图分类的模块

Route different intents to specialized handlers:
python
from typing import Literal

class ClassifyIntent(dspy.Signature):
    """Classify the user's intent from their message."""
    conversation_history: str = dspy.InputField()
    user_message: str = dspy.InputField()
    intent: Literal["question", "complaint", "request", "greeting", "goodbye"] = dspy.OutputField()

class ChatBotWithRouting(dspy.Module):
    def __init__(self):
        self.classify = dspy.Predict(ClassifyIntent)
        self.respond_question = dspy.ChainOfThought(AnswerQuestion)
        self.respond_complaint = dspy.ChainOfThought(HandleComplaint)
        self.respond_request = dspy.ChainOfThought(HandleRequest)
        self.respond_greeting = dspy.Predict(Greeting)

    def forward(self, conversation_history, context, user_message):
        intent = self.classify(
            conversation_history=conversation_history,
            user_message=user_message,
        ).intent

        handler = {
            "question": self.respond_question,
            "complaint": self.respond_complaint,
            "request": self.respond_request,
            "greeting": self.respond_greeting,
        }.get(intent, self.respond_question)

        return handler(
            conversation_history=conversation_history,
            context=context,
            user_message=user_message,
        )
将不同意图路由至对应的专用处理器：
python
from typing import Literal

class ClassifyIntent(dspy.Signature):
    """Classify the user's intent from their message."""
    conversation_history: str = dspy.InputField()
    user_message: str = dspy.InputField()
    intent: Literal["question", "complaint", "request", "greeting", "goodbye"] = dspy.OutputField()

class ChatBotWithRouting(dspy.Module):
    def __init__(self):
        self.classify = dspy.Predict(ClassifyIntent)
        self.respond_question = dspy.ChainOfThought(AnswerQuestion)
        self.respond_complaint = dspy.ChainOfThought(HandleComplaint)
        self.respond_request = dspy.ChainOfThought(HandleRequest)
        self.respond_greeting = dspy.Predict(Greeting)

    def forward(self, conversation_history, context, user_message):
        intent = self.classify(
            conversation_history=conversation_history,
            user_message=user_message,
        ).intent

        handler = {
            "question": self.respond_question,
            "complaint": self.respond_complaint,
            "request": self.respond_request,
            "greeting": self.respond_greeting,
        }.get(intent, self.respond_question)

        return handler(
            conversation_history=conversation_history,
            context=context,
            user_message=user_message,
        )

Step 3: Add conversation state (LangGraph)

步骤3：添加对话状态管理（LangGraph）

LangGraph manages the conversation flow — what state the bot is in, when to transition, and when to escalate.
LangGraph负责管理对话流程——包括机器人当前所处的状态、状态切换时机以及何时需要升级处理。

Define conversation state

定义对话状态

python
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
import operator

class ConversationState(TypedDict):
    messages: Annotated[list[dict], operator.add]  # full message history
    current_intent: str
    context: str           # retrieved docs/data for current turn
    escalate: bool         # whether to hand off to a human
    resolved: bool         # whether the issue is resolved
    turn_count: int
python
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
import operator

class ConversationState(TypedDict):
    messages: Annotated[list[dict], operator.add]  # full message history
    current_intent: str
    context: str           # retrieved docs/data for current turn
    escalate: bool         # whether to hand off to a human
    resolved: bool         # whether the issue is resolved
    turn_count: int

Build the conversation graph

构建对话图

python
import dspy
python
import dspy

Initialize DSPy modules

Initialize DSPy modules

classifier = dspy.Predict(ClassifyIntent) responder = dspy.ChainOfThought(ChatResponse)
def classify_node(state: ConversationState) -> dict: """Classify the user's intent.""" history = format_history(state["messages"][:-1]) user_msg = state["messages"][-1]["content"] result = classifier(conversation_history=history, user_message=user_msg) return {"current_intent": result.intent}
def retrieve_node(state: ConversationState) -> dict: """Retrieve relevant docs for the current message.""" user_msg = state["messages"][-1]["content"] # Your retrieval logic here (see /ai-searching-docs) docs = retrieve_relevant_docs(user_msg) return {"context": "\n".join(docs)}
def respond_node(state: ConversationState) -> dict: """Generate a response using DSPy.""" history = format_history(state["messages"][:-1]) user_msg = state["messages"][-1]["content"] result = responder( conversation_history=history, context=state["context"], user_message=user_msg, ) return { "messages": [{"role": "assistant", "content": result.response}], "turn_count": state["turn_count"] + 1, }
def check_escalation(state: ConversationState) -> dict: """Decide if this needs human handoff.""" should_escalate = ( state["current_intent"] == "complaint" and state["turn_count"] > 3 ) return {"escalate": should_escalate}
def format_history(messages: list[dict]) -> str: return "\n".join(f"{m['role']}: {m['content']}" for m in messages[-10:])
classifier = dspy.Predict(ClassifyIntent) responder = dspy.ChainOfThought(ChatResponse)
def classify_node(state: ConversationState) -> dict: """Classify the user's intent.""" history = format_history(state["messages"][:-1]) user_msg = state["messages"][-1]["content"] result = classifier(conversation_history=history, user_message=user_msg) return {"current_intent": result.intent}
def retrieve_node(state: ConversationState) -> dict: """Retrieve relevant docs for the current message.""" user_msg = state["messages"][-1]["content"] # Your retrieval logic here (see /ai-searching-docs) docs = retrieve_relevant_docs(user_msg) return {"context": "\n".join(docs)}
def respond_node(state: ConversationState) -> dict: """Generate a response using DSPy.""" history = format_history(state["messages"][:-1]) user_msg = state["messages"][-1]["content"] result = responder( conversation_history=history, context=state["context"], user_message=user_msg, ) return { "messages": [{"role": "assistant", "content": result.response}], "turn_count": state["turn_count"] + 1, }
def check_escalation(state: ConversationState) -> dict: """Decide if this needs human handoff.""" should_escalate = ( state["current_intent"] == "complaint" and state["turn_count"] > 3 ) return {"escalate": should_escalate}
def format_history(messages: list[dict]) -> str: return "\n".join(f"{m['role']}: {m['content']}" for m in messages[-10:])

Build the graph

Build the graph

graph = StateGraph(ConversationState) graph.add_node("classify", classify_node) graph.add_node("retrieve", retrieve_node) graph.add_node("respond", respond_node) graph.add_node("check_escalation", check_escalation)
graph.add_edge(START, "classify") graph.add_edge("classify", "retrieve") graph.add_edge("retrieve", "respond") graph.add_edge("respond", "check_escalation")
def route_after_escalation_check(state: ConversationState) -> str: if state["escalate"]: return "escalate" return "done"
graph.add_conditional_edges( "check_escalation", route_after_escalation_check, {"escalate": END, "done": END}, )
app = graph.compile()
undefined
graph = StateGraph(ConversationState) graph.add_node("classify", classify_node) graph.add_node("retrieve", retrieve_node) graph.add_node("respond", respond_node) graph.add_node("check_escalation", check_escalation)
graph.add_edge(START, "classify") graph.add_edge("classify", "retrieve") graph.add_edge("retrieve", "respond") graph.add_edge("respond", "check_escalation")
def route_after_escalation_check(state: ConversationState) -> str: if state["escalate"]: return "escalate" return "done"
graph.add_conditional_edges( "check_escalation", route_after_escalation_check, {"escalate": END, "done": END}, )
app = graph.compile()
undefined

Run a conversation turn

运行单轮对话

python
result = app.invoke({
    "messages": [{"role": "user", "content": "How do I reset my password?"}],
    "current_intent": "",
    "context": "",
    "escalate": False,
    "resolved": False,
    "turn_count": 0,
})
print(result["messages"][-1]["content"])
python
result = app.invoke({
    "messages": [{"role": "user", "content": "How do I reset my password?"}],
    "current_intent": "",
    "context": "",
    "escalate": False,
    "resolved": False,
    "turn_count": 0,
})
print(result["messages"][-1]["content"])

Step 4: Add memory

步骤4：添加记忆功能

Session memory with checkpointing

带检查点的会话记忆

LangGraph's checkpointer persists conversation state across requests:
python
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)
LangGraph的检查点功能可在多次请求间持久化对话状态：
python
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

Each user session gets a unique thread_id

Each user session gets a unique thread_id

config = {"configurable": {"thread_id": "user-abc-123"}}
config = {"configurable": {"thread_id": "user-abc-123"}}

Turn 1

Turn 1

result = app.invoke( {"messages": [{"role": "user", "content": "Hi, I need help with billing"}], "current_intent": "", "context": "", "escalate": False, "resolved": False, "turn_count": 0}, config=config, )
result = app.invoke( {"messages": [{"role": "user", "content": "Hi, I need help with billing"}], "current_intent": "", "context": "", "escalate": False, "resolved": False, "turn_count": 0}, config=config, )

Turn 2 — state is preserved, the bot remembers the conversation

Turn 2 — state is preserved, the bot remembers the conversation

result = app.invoke( {"messages": [{"role": "user", "content": "I was charged twice last month"}]}, config=config, )

For production, use a persistent backend:

```python
from langgraph.checkpoint.postgres import PostgresSaver

checkpointer = PostgresSaver(conn_string="postgresql://user:pass@localhost/chatbot")
app = graph.compile(checkpointer=checkpointer)
result = app.invoke( {"messages": [{"role": "user", "content": "I was charged twice last month"}]}, config=config, )

生产环境中，可使用持久化后端：

```python
from langgraph.checkpoint.postgres import PostgresSaver

checkpointer = PostgresSaver(conn_string="postgresql://user:pass@localhost/chatbot")
app = graph.compile(checkpointer=checkpointer)

Conversation summary for long chats

长对话总结

When conversations get long, summarize older messages to stay within token limits:
python
class SummarizeConversation(dspy.Signature):
    """Summarize the conversation so far, preserving key details."""
    conversation: str = dspy.InputField()
    summary: str = dspy.OutputField(desc="Concise summary of the conversation so far")

summarizer = dspy.Predict(SummarizeConversation)

def maybe_summarize(state: ConversationState) -> dict:
    """Summarize if conversation is getting long."""
    if len(state["messages"]) > 20:
        history = format_history(state["messages"][:-5])
        summary = summarizer(conversation=history).summary
        # Keep summary + last 5 messages
        return {
            "messages": [
                {"role": "system", "content": f"Summary of earlier conversation: {summary}"},
                *state["messages"][-5:],
            ]
        }
    return {}
当对话过长时，总结早期消息以控制token用量：
python
class SummarizeConversation(dspy.Signature):
    """Summarize the conversation so far, preserving key details."""
    conversation: str = dspy.InputField()
    summary: str = dspy.OutputField(desc="Concise summary of the conversation so far")

summarizer = dspy.Predict(SummarizeConversation)

def maybe_summarize(state: ConversationState) -> dict:
    """Summarize if conversation is getting long."""
    if len(state["messages"]) > 20:
        history = format_history(state["messages"][:-5])
        summary = summarizer(conversation=history).summary
        # Keep summary + last 5 messages
        return {
            "messages": [
                {"role": "system", "content": f"Summary of earlier conversation: {summary}"},
                *state["messages"][-5:],
            ]
        }
    return {}

Step 5: Ground responses in docs

步骤5：基于文档生成响应

Retrieve relevant documents each turn to keep responses factual.
python
class DocGroundedResponse(dspy.Signature):
    """Answer the user's question based on the provided documentation.
    Only use information from the docs. If the docs don't cover it, say so."""
    conversation_history: str = dspy.InputField()
    docs: list[str] = dspy.InputField(desc="Relevant documentation passages")
    user_message: str = dspy.InputField()
    response: str = dspy.OutputField()

class GroundedChatBot(dspy.Module):
    def __init__(self, retriever):
        self.retriever = retriever
        self.respond = dspy.ChainOfThought(DocGroundedResponse)

    def forward(self, conversation_history, user_message):
        # Retrieve docs relevant to the current message
        docs = self.retriever(user_message).passages
        return self.respond(
            conversation_history=conversation_history,
            docs=docs,
            user_message=user_message,
        )
See 
/ai-searching-docs
for setting up retrievers and vector stores, including loading data from PDFs, Notion, and other sources with LangChain document loaders.
每轮对话都检索相关文档，确保响应真实可信。
python
class DocGroundedResponse(dspy.Signature):
    """Answer the user's question based on the provided documentation.
    Only use information from the docs. If the docs don't cover it, say so."""
    conversation_history: str = dspy.InputField()
    docs: list[str] = dspy.InputField(desc="Relevant documentation passages")
    user_message: str = dspy.InputField()
    response: str = dspy.OutputField()

class GroundedChatBot(dspy.Module):
    def __init__(self, retriever):
        self.retriever = retriever
        self.respond = dspy.ChainOfThought(DocGroundedResponse)

    def forward(self, conversation_history, user_message):
        # Retrieve docs relevant to the current message
        docs = self.retriever(user_message).passages
        return self.respond(
            conversation_history=conversation_history,
            docs=docs,
            user_message=user_message,
        )
如需设置检索器和向量存储，包括使用LangChain文档加载器从PDF、Notion等来源加载数据，请参考 
/ai-searching-docs
。

Step 6: Add guardrails

步骤6：添加安全防护

Response quality with DSPy assertions

基于DSPy断言的响应质量管控

python
class GuardedChatBot(dspy.Module):
    def __init__(self, retriever):
        self.retriever = retriever
        self.respond = dspy.ChainOfThought(DocGroundedResponse)

    def forward(self, conversation_history, user_message):
        docs = self.retriever(user_message).passages
        result = self.respond(
            conversation_history=conversation_history,
            docs=docs,
            user_message=user_message,
        )

        # Guardrails
        dspy.Suggest(
            len(result.response.split()) < 200,
            "Keep responses concise — under 200 words",
        )
        dspy.Suggest(
            not any(word in result.response.lower() for word in ["obviously", "clearly", "simply"]),
            "Avoid condescending language",
        )
        dspy.Assert(
            "I am an AI" not in result.response,
            "Don't break character with meta-statements",
        )

        return result
python
class GuardedChatBot(dspy.Module):
    def __init__(self, retriever):
        self.retriever = retriever
        self.respond = dspy.ChainOfThought(DocGroundedResponse)

    def forward(self, conversation_history, user_message):
        docs = self.retriever(user_message).passages
        result = self.respond(
            conversation_history=conversation_history,
            docs=docs,
            user_message=user_message,
        )

        # Guardrails
        dspy.Suggest(
            len(result.response.split()) < 200,
            "Keep responses concise — under 200 words",
        )
        dspy.Suggest(
            not any(word in result.response.lower() for word in ["obviously", "clearly", "simply"]),
            "Avoid condescending language",
        )
        dspy.Assert(
            "I am an AI" not in result.response,
            "Don't break character with meta-statements",
        )

        return result

Human-in-the-loop for sensitive actions

敏感操作的人工介入机制

Use LangGraph's interrupt to pause before the bot takes real actions:
python
app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["execute_refund", "cancel_account"],  # pause here
)
使用LangGraph的中断功能，在机器人执行实际操作前暂停：
python
app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["execute_refund", "cancel_account"],  # pause here
)

Bot runs until it reaches a sensitive action

Bot runs until it reaches a sensitive action

result = app.invoke(input_state, config)
result = app.invoke(input_state, config)

Human agent reviews the proposed action

Human agent reviews the proposed action

If approved, resume:

If approved, resume:

result = app.invoke(None, config) # continues from checkpoint
undefined
result = app.invoke(None, config) # continues from checkpoint
undefined

Step 7: Optimize and evaluate

步骤7：优化与评估

Conversation-level metrics

对话级指标

python
def chatbot_metric(example, prediction, trace=None):
    """Score a single conversation turn."""
    judge = dspy.Predict(JudgeTurn)
    result = judge(
        user_message=example.user_message,
        expected_response=example.response,
        actual_response=prediction.response,
        conversation_history=example.conversation_history,
    )
    return result.is_good

class JudgeTurn(dspy.Signature):
    """Judge if the chatbot response is helpful, accurate, and on-topic."""
    user_message: str = dspy.InputField()
    expected_response: str = dspy.InputField()
    actual_response: str = dspy.InputField()
    conversation_history: str = dspy.InputField()
    is_good: bool = dspy.OutputField()
python
def chatbot_metric(example, prediction, trace=None):
    """Score a single conversation turn."""
    judge = dspy.Predict(JudgeTurn)
    result = judge(
        user_message=example.user_message,
        expected_response=example.response,
        actual_response=prediction.response,
        conversation_history=example.conversation_history,
    )
    return result.is_good

class JudgeTurn(dspy.Signature):
    """Judge if the chatbot response is helpful, accurate, and on-topic."""
    user_message: str = dspy.InputField()
    expected_response: str = dspy.InputField()
    actual_response: str = dspy.InputField()
    conversation_history: str = dspy.InputField()
    is_good: bool = dspy.OutputField()

Build a training set from real conversations

基于真实对话构建训练集

python
trainset = []
for convo in real_conversations:
    for turn in convo["turns"]:
        trainset.append(
            dspy.Example(
                conversation_history=turn["history"],
                user_message=turn["user_message"],
                context=turn["context"],
                response=turn["response"],
            ).with_inputs("conversation_history", "user_message", "context")
        )
python
trainset = []
for convo in real_conversations:
    for turn in convo["turns"]:
        trainset.append(
            dspy.Example(
                conversation_history=turn["history"],
                user_message=turn["user_message"],
                context=turn["context"],
                response=turn["response"],
            ).with_inputs("conversation_history", "user_message", "context")
        )

Optimize

优化模型

python
optimizer = dspy.MIPROv2(metric=chatbot_metric, auto="medium")
optimized_bot = optimizer.compile(chatbot, trainset=trainset)
python
optimizer = dspy.MIPROv2(metric=chatbot_metric, auto="medium")
optimized_bot = optimizer.compile(chatbot, trainset=trainset)

Save optimized prompts

Save optimized prompts

optimized_bot.save("chatbot_optimized.json")
undefined
optimized_bot.save("chatbot_optimized.json")
undefined

Key patterns

核心模式

  • DSPy for response generation, LangGraph for flow control — DSPy modules handle what the bot says; LangGraph handles conversation state and routing
  • Checkpointing is your memory — use LangGraph's checkpointer so conversations persist across HTTP requests
  • Retrieve every turn — don't assume context from earlier turns is still relevant; re-retrieve each time
  • Summarize long conversations — once past ~20 messages, summarize older context to stay within token limits
  • Classify intent early — knowing the user's intent lets you route to specialized handlers
  • Interrupt before real actions — use LangGraph's 
    interrupt_before
    so humans approve refunds, cancellations, etc.
  • Optimize on real conversations — collect actual chat logs to build training data for DSPy optimization
  • DSPy负责响应生成，LangGraph负责流程控制 —— DSPy模块处理机器人的回复内容；LangGraph处理对话状态与路由
  • 检查点即记忆 —— 使用LangGraph的检查点功能，让对话在HTTP请求间保持持续
  • 每轮都检索 —— 不要假设早期对话的上下文仍然相关，每轮对话都重新检索信息
  • 长对话总结 —— 当对话超过约20条消息时，总结早期上下文以控制token用量
  • 提前分类意图 —— 了解用户意图后，可将对话路由至专用处理器
  • 敏感操作前中断 —— 使用LangGraph的
    interrupt_before
    功能，让人工审核退款、账户注销等操作
  • 基于真实对话优化 —— 收集实际聊天记录，为DSPy优化构建训练数据

Additional resources

额外资源

  • For worked examples (support bot, FAQ assistant), see examples.md
  • For the LangChain/LangGraph API reference, see 
    docs/langchain-langgraph-reference.md
  • Need to search docs for grounding? Use 
    /ai-searching-docs
  • Need the bot to take actions (call APIs, tools)? Use 
    /ai-taking-actions
  • Building multiple bots that work together? Use 
    /ai-coordinating-agents
  • Next: 
    /ai-improving-accuracy
    to measure and improve your chatbot
  • 如需完整示例（支持机器人、FAQ助手），请查看 examples.md
  • 如需LangChain/LangGraph API参考文档，请查看 
    docs/langchain-langgraph-reference.md
  • 如需基于文档的搜索功能，请使用 
    /ai-searching-docs
  • 如需让机器人执行操作（调用API、工具），请使用 
    /ai-taking-actions
  • 如需构建协同工作的多机器人系统，请使用 
    /ai-coordinating-agents
  • 下一步：使用 
    /ai-improving-accuracy
    测量并提升聊天机器人的准确度