使用LangChain构建聊天机器人

LangChain让构建AI应用变得简单。本教程将教你如何使用LangChain创建一个智能聊天机器人。

什么是LangChain？

LangChain是一个用于开发由语言模型驱动的应用程序的框架。

核心优势

模块化设计: 可组合的组件
链式调用: 连接多个操作
记忆管理: 维护对话上下文
工具集成: 连接外部API和数据库

环境设置

安装依赖

# 创建虚拟环境
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# 安装LangChain和相关库
pip install langchain
pip install langchain-openai
pip install python-dotenv
pip install streamlit  # 用于Web界面

配置API密钥

# .env 文件
OPENAI_API_KEY=your_api_key_here

# config.py
import os
from dotenv import load_dotenv

load_dotenv()

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")

if not OPENAI_API_KEY:
    raise ValueError("请设置OPENAI_API_KEY环境变量")

基础聊天机器人

最简单的实现

from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage, SystemMessage

# 初始化模型
llm = ChatOpenAI(
    model="gpt-4",
    temperature=0.7,
    max_tokens=500
)

# 发送消息
messages = [
    SystemMessage(content="你是一个友好的AI助手。"),
    HumanMessage(content="你好！")
]

response = llm.invoke(messages)
print(response.content)

添加对话历史

from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain

# 创建记忆
memory = ConversationBufferMemory()

# 创建对话链
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

# 多轮对话
print(conversation.predict(input="我叫张三"))
print(conversation.predict(input="我叫什么名字？"))
# 输出: 你叫张三

高级记忆管理

对话缓冲窗口记忆

from langchain.memory import ConversationBufferWindowMemory

# 只保留最近5轮对话
memory = ConversationBufferWindowMemory(k=5)

conversation = ConversationChain(
    llm=llm,
    memory=memory
)

对话摘要记忆

from langchain.memory import ConversationSummaryMemory

# 自动总结历史对话
memory = ConversationSummaryMemory(llm=llm)

conversation = ConversationChain(
    llm=llm,
    memory=memory
)

实体记忆

from langchain.memory import ConversationEntityMemory

# 记住对话中的实体信息
memory = ConversationEntityMemory(llm=llm)

conversation = ConversationChain(
    llm=llm,
    memory=memory
)

# 机器人会记住用户提到的人名、地点等实体

提示词模板

基础模板

from langchain.prompts import PromptTemplate

# 定义模板
template = """
你是一个{role}。
用户问题: {question}
请提供专业的回答。
"""

prompt = PromptTemplate(
    input_variables=["role", "question"],
    template=template
)

# 使用模板
formatted_prompt = prompt.format(
    role="Python专家",
    question="如何优化代码性能？"
)

聊天提示词模板

from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个{role}，擅长{expertise}。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])

# 使用
chain = prompt | llm
response = chain.invoke({
    "role": "编程导师",
    "expertise": "Python和机器学习",
    "history": [],
    "input": "如何学习深度学习？"
})

工具集成

搜索工具

from langchain.agents import load_tools, initialize_agent, AgentType

# 加载工具
tools = load_tools(["serpapi", "llm-math"], llm=llm)

# 创建代理
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# 使用代理
response = agent.run("2024年世界杯冠军是谁？")

自定义工具

from langchain.tools import Tool
from langchain.agents import initialize_agent

def get_weather(location: str) -> str:
    """获取天气信息"""
    # 实际应用中调用天气API
    return f"{location}的天气是晴天，温度25度"

def calculate_age(birth_year: int) -> int:
    """计算年龄"""
    from datetime import datetime
    current_year = datetime.now().year
    return current_year - birth_year

# 定义工具
tools = [
    Tool(
        name="Weather",
        func=get_weather,
        description="获取指定地点的天气信息。输入应该是城市名称。"
    ),
    Tool(
        name="AgeCalculator",
        func=calculate_age,
        description="根据出生年份计算年龄。输入应该是四位数的年份。"
    )
]

# 创建代理
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# 测试
print(agent.run("北京的天气怎么样？"))
print(agent.run("1990年出生的人现在多大？"))

文档问答系统

加载和处理文档

from langchain.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS

# 加载文档
loader = TextLoader("document.txt", encoding="utf-8")
documents = loader.load()

# 分割文档
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
texts = text_splitter.split_documents(documents)

# 创建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_documents(texts, embeddings)

构建问答链

from langchain.chains import RetrievalQA

# 创建问答链
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(search_kwargs={"k": 3})
)

# 提问
question = "文档的主要内容是什么？"
answer = qa_chain.run(question)
print(answer)

对话式文档问答

from langchain.chains import ConversationalRetrievalChain

# 创建对话式问答链
qa_chain = ConversationalRetrievalChain.from_llm(
    llm=llm,
    retriever=vectorstore.as_retriever(),
    memory=ConversationBufferMemory(
        memory_key="chat_history",
        return_messages=True
    )
)

# 多轮问答
result = qa_chain({"question": "文档讲了什么？"})
print(result["answer"])

result = qa_chain({"question": "能详细说说吗？"})
print(result["answer"])

完整的聊天机器人类

from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferWindowMemory
from langchain.chains import ConversationChain
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
import json
from datetime import datetime

class ChatBot:
    """智能聊天机器人"""

    def __init__(self, model="gpt-4", temperature=0.7):
        self.llm = ChatOpenAI(
            model=model,
            temperature=temperature
        )
        self.memory = ConversationBufferWindowMemory(
            k=10,
            return_messages=True
        )
        self.conversation = self._create_conversation()

    def _create_conversation(self):
        """创建对话链"""
        prompt = ChatPromptTemplate.from_messages([
            ("system", """你是一个友好、专业的AI助手。
            当前时间: {current_time}

            请遵循以下原则:
            1. 提供准确、有帮助的回答
            2. 如果不确定，诚实地说明
            3. 保持对话自然流畅
            4. 记住之前的对话内容
            """),
            MessagesPlaceholder(variable_name="history"),
            ("human", "{input}")
        ])

        chain = ConversationChain(
            llm=self.llm,
            memory=self.memory,
            prompt=prompt,
            verbose=False
        )
        return chain

    def chat(self, message: str) -> str:
        """发送消息并获取回复"""
        try:
            response = self.conversation.predict(
                input=message,
                current_time=datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            )
            return response
        except Exception as e:
            return f"抱歉，发生错误: {str(e)}"

    def clear_history(self):
        """清除对话历史"""
        self.memory.clear()

    def get_history(self):
        """获取对话历史"""
        return self.memory.load_memory_variables({})

    def save_conversation(self, filename: str):
        """保存对话到文件"""
        history = self.get_history()
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(history, f, ensure_ascii=False, indent=2)

# 使用示例
if __name__ == "__main__":
    bot = ChatBot()

    print("聊天机器人已启动！输入 'quit' 退出。\n")

    while True:
        user_input = input("你: ")

        if user_input.lower() in ['quit', 'exit', '退出']:
            print("再见！")
            break

        if user_input.lower() == 'clear':
            bot.clear_history()
            print("对话历史已清除。")
            continue

        response = bot.chat(user_input)
        print(f"机器人: {response}\n")

Streamlit Web界面

import streamlit as st
from chatbot import ChatBot

# 页面配置
st.set_page_config(
    page_title="AI聊天机器人",
    page_icon="🤖",
    layout="wide"
)

# 初始化
if 'bot' not in st.session_state:
    st.session_state.bot = ChatBot()
if 'messages' not in st.session_state:
    st.session_state.messages = []

# 标题
st.title("🤖 AI聊天机器人")

# 侧边栏
with st.sidebar:
    st.header("设置")

    if st.button("清除对话"):
        st.session_state.bot.clear_history()
        st.session_state.messages = []
        st.success("对话已清除！")

    st.markdown("---")
    st.markdown("### 关于")
    st.markdown("基于LangChain和OpenAI构建的智能聊天机器人。")

# 显示对话历史
for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

# 用户输入
if prompt := st.chat_input("输入你的消息..."):
    # 显示用户消息
    st.session_state.messages.append({"role": "user", "content": prompt})
    with st.chat_message("user"):
        st.markdown(prompt)

    # 获取机器人回复
    with st.chat_message("assistant"):
        with st.spinner("思考中..."):
            response = st.session_state.bot.chat(prompt)
            st.markdown(response)

    # 保存机器人回复
    st.session_state.messages.append({"role": "assistant", "content": response})

运行Web应用:

streamlit run app.py

流式响应

from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

# 启用流式输出
llm = ChatOpenAI(
    model="gpt-4",
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()]
)

# 自定义流式回调
from langchain.callbacks.base import BaseCallbackHandler

class StreamHandler(BaseCallbackHandler):
    def __init__(self):
        self.text = ""

    def on_llm_new_token(self, token: str, **kwargs):
        self.text += token
        print(token, end="", flush=True)

# 使用
handler = StreamHandler()
llm = ChatOpenAI(
    model="gpt-4",
    streaming=True,
    callbacks=[handler]
)

错误处理和重试

from langchain.llms import OpenAI
from langchain.callbacks import get_openai_callback
import time

def chat_with_retry(bot, message, max_retries=3):
    """带重试的聊天"""
    for attempt in range(max_retries):
        try:
            with get_openai_callback() as cb:
                response = bot.chat(message)
                print(f"Token使用: {cb.total_tokens}")
                return response
        except Exception as e:
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt
                print(f"错误: {e}. {wait_time}秒后重试...")
                time.sleep(wait_time)
            else:
                return f"抱歉，服务暂时不可用: {str(e)}"

性能优化

缓存响应

from langchain.cache import InMemoryCache
from langchain.globals import set_llm_cache

# 启用缓存
set_llm_cache(InMemoryCache())

# 相同的问题会直接返回缓存结果

批量处理

# 批量生成响应
messages_batch = [
    [HumanMessage(content="你好")],
    [HumanMessage(content="天气怎么样？")],
    [HumanMessage(content="推荐一本书")]
]

responses = llm.generate(messages_batch)
for response in responses.generations:
    print(response[0].text)

部署到生产环境

Docker部署

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8501

CMD ["streamlit", "run", "app.py", "--server.port=8501", "--server.address=0.0.0.0"]

# 构建和运行
docker build -t chatbot .
docker run -p 8501:8501 -e OPENAI_API_KEY=your_key chatbot

环境变量管理

# config.py
import os

class Config:
    OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
    MODEL_NAME = os.getenv("MODEL_NAME", "gpt-4")
    TEMPERATURE = float(os.getenv("TEMPERATURE", "0.7"))
    MAX_TOKENS = int(os.getenv("MAX_TOKENS", "500"))
    MEMORY_SIZE = int(os.getenv("MEMORY_SIZE", "10"))

    @classmethod
    def validate(cls):
        if not cls.OPENAI_API_KEY:
            raise ValueError("OPENAI_API_KEY is required")

最佳实践

1. 提示词工程

# 好的提示词
good_prompt = """
你是一个专业的客服助手。

职责:
- 回答产品相关问题
- 处理客户投诉
- 提供技术支持

风格:
- 友好、专业
- 简洁明了
- 富有同理心

限制:
- 不讨论竞争对手
- 不承诺无法兑现的事情
- 遇到不确定的问题，引导用户联系人工客服
"""

2. 安全性

def sanitize_input(user_input: str) -> str:
    """清理用户输入"""
    # 移除潜在的注入攻击
    dangerous_patterns = ['<script>', 'javascript:', 'onerror=']
    for pattern in dangerous_patterns:
        user_input = user_input.replace(pattern, '')
    return user_input.strip()

def validate_response(response: str) -> bool:
    """验证响应内容"""
    # 检查是否包含敏感信息
    sensitive_keywords = ['密码', 'API密钥', '信用卡']
    return not any(keyword in response for keyword in sensitive_keywords)

3. 监控和日志

import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('chatbot.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

def chat_with_logging(bot, message):
    logger.info(f"User input: {message}")
    try:
        response = bot.chat(message)
        logger.info(f"Bot response: {response}")
        return response
    except Exception as e:
        logger.error(f"Error: {str(e)}", exc_info=True)
        raise

总结

使用LangChain构建聊天机器人的关键步骤:

设置环境和API密钥
选择合适的记忆管理策略
设计有效的提示词模板
集成必要的工具和功能
实现错误处理和重试机制
优化性能和用户体验
安全部署到生产环境

LangChain让AI应用开发变得简单高效，现在就开始构建你的聊天机器人吧！

使用LangChain构建聊天机器人

分享这篇文章