对话系统实战

6.1 项目一：智能聊天机器人

构建一个带记忆功能的智能聊天机器人，支持多会话、上下文理解和流式输出。

6.1.1 完整代码

"""
智能聊天机器人 - 带记忆和多会话支持
"""

import os
from datetime import datetime
from typing import Optional, Dict, List
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_core.runnables.history import RunnableWithMessageHistory

# ==================== 配置 ====================
os.environ["OPENAI_API_KEY"] = "your-api-key"

# ==================== 初始化 ====================
llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0.8,
    streaming=True  # 启用流式输出
)

# 系统提示
SYSTEM_PROMPT = """你是一个友善、有帮助的 AI 助手。
- 回答问题简洁明了
- 可以进行轻松的对话
- 如果用户自我介绍，记住了并在后续对话中使用
- 使用中文回答"""

# 创建提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", SYSTEM_PROMPT),
    MessagesPlaceholder("history"),
    ("human", "{message}"),
])

# 创建链
chain = prompt | llm

# ==================== 会话管理 ====================
class ChatBot:
    def __init__(self):
        self.store: Dict[str, InMemoryChatMessageHistory] = {}

    def get_history(self, session_id: str) -> InMemoryChatMessageHistory:
        """获取或创建会话历史"""
        if session_id not in self.store:
            self.store[session_id] = InMemoryChatMessageHistory()
        return self.store[session_id]

    def create_chain_with_history(self, session_id: str):
        """为指定会话创建带历史的链"""
        return RunnableWithMessageHistory(
            chain,
            self.get_history,
            input_messages_key="message",
            history_messages_key="history",
        )

    def chat(self, session_id: str, message: str) -> str:
        """发送消息并获取回复"""
        chain_with_history = self.create_chain_with_history(session_id)
        response = chain_with_history.invoke(
            {"message": message},
            config={"configurable": {"session_id": session_id}}
        )
        return response.content

    def get_history_count(self, session_id: str) -> int:
        """获取会话消息数"""
        if session_id in self.store:
            return len(self.store[session_id].messages)
        return 0

    def clear_history(self, session_id: str):
        """清除会话历史"""
        if session_id in self.store:
            self.store[session_id].clear()

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

    # 创建会话
    session_id = "user_001"

    print("=" * 50)
    print("🤖 智能聊天机器人 (输入 'quit' 退出)")
    print("=" * 50)

    while True:
        user_input = input("\n你: ").strip()
        if user_input.lower() == "quit":
            print("再见！")
            break

        response = bot.chat(session_id, user_input)
        print(f"\nAI: {response}")

        # 打印当前消息数
        count = bot.get_history_count(session_id)
        print(f"[会话消息数: {count}]")

6.1.2 进阶功能：添加摘要记忆

"""
带摘要记忆的聊天机器人 - 解决长对话问题
"""

from langchain.memory import ConversationSummaryMemory
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder

llm = ChatOpenAI(model="gpt-4o-mini")

# 使用摘要记忆
memory = ConversationSummaryMemory(
    llm=llm,
    memory_key="history",
    return_messages=True,
    max_token_limit=500  # 超过限制自动摘要
)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的助手。已知信息：{history}"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{message}"),
])

# 对话
memory.save_context({"input": "我叫李明"}, {"output": "你好李明！"})
memory.save_context({"input": "我是一名软件工程师"}, {"output": "好的，工程师李明！"})

# 获取摘要
summary = memory.load_memory_variables({})["history"]
print("记忆摘要:", summary)

6.2 项目二：PDF 文档问答

构建一个基于 RAG 的 PDF 文档问答系统，可以上传 PDF 并针对文档内容提问。

6.2.1 完整代码

"""
PDF 文档问答系统
pip install langchain langchain-openai langchain-community pypdf faiss-cpu
"""

import os
from typing import List, Optional
from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.documents import Document

# ==================== 配置 ====================
class PDFQAConfig:
    CHUNK_SIZE = 500
    CHUNK_OVERLAP = 50
    RETRIEVAL_K = 3  # 检索的文档块数
    MODEL_NAME = "gpt-4o-mini"
    EMBEDDING_MODEL = "text-embedding-3-small"

# ==================== PDF 处理器 ====================
class PDFProcessor:
    def __init__(self):
        self.embeddings = OpenAIEmbeddings(
            model=PDFQAConfig.EMBEDDING_MODEL
        )
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=PDFQAConfig.CHUNK_SIZE,
            chunk_overlap=PDFQAConfig.CHUNK_OVERLAP,
            length_function=len,
        )

    def load_and_split(self, pdf_path: str) -> List[Document]:
        """加载并分割 PDF"""
        loader = PyPDFLoader(pdf_path)
        pages = loader.load()
        chunks = self.text_splitter.split_documents(pages)
        print(f"📄 已加载 {len(pages)} 页，生成 {len(chunks)} 个文本块")
        return chunks

    def create_vectorstore(self, chunks: List[Document]) -> FAISS:
        """创建向量存储"""
        vectorstore = FAISS.from_documents(
            documents=chunks,
            embedding=self.embeddings
        )
        print("✅ 向量数据库创建完成")
        return vectorstore

    def load_vectorstore(self, path: str) -> FAISS:
        """加载已有向量存储"""
        return FAISS.load_local(
            path,
            self.embeddings,
            allow_dangerous_deserialization=True
        )

    def save_vectorstore(self, vectorstore: FAISS, path: str):
        """保存向量存储"""
        vectorstore.save_local(path)
        print(f"💾 已保存到 {path}")

# ==================== RAG 问答 ====================
class PDFQA:
    def __init__(self, vectorstore: FAISS):
        self.vectorstore = vectorstore
        self.retriever = vectorstore.as_retriever(
            search_kwargs={"k": PDFQAConfig.RETRIEVAL_K}
        )
        self.llm = ChatOpenAI(
            model=PDFQAConfig.MODEL_NAME,
            temperature=0
        )

        # RAG 提示模板
        self.prompt = ChatPromptTemplate.from_template("""根据以下上下文回答问题。
如果上下文中没有相关信息，请回答"我没有找到相关信息"。

上下文：
{context}

问题：{question}

回答：""")

        # 构建链
        self.chain = (
            {
                "context": self.retriever | self._format_docs,
                "question": lambda x: x
            }
            | self.prompt
            | self.llm
            | StrOutputParser()
        )

    @staticmethod
    def _format_docs(docs: List[Document]) -> str:
        """格式化检索到的文档"""
        return "\n\n".join([
            f"[来源: 第{doc.metadata.get('page', '?')}页]\n{doc.page_content}"
            for doc in docs
        ])

    def ask(self, question: str) -> tuple[str, List[Document]]:
        """提问"""
        answer = self.chain.invoke(question)

        # 获取源文档
        docs = self.vectorstore.similarity_search(
            question,
            k=PDFQAConfig.RETRIEVAL_K
        )

        return answer, docs

# ==================== 使用示例 ====================
if __name__ == "__main__":
    # 初始化
    processor = PDFProcessor()

    # 处理 PDF（首次运行）
    chunks = processor.load_and_split("document.pdf")
    vectorstore = processor.create_vectorstore(chunks)
    processor.save_vectorstore(vectorstore, "pdf_vectorstore")

    # 加载已有向量库（后续运行）
    # vectorstore = processor.load_vectorstore("pdf_vectorstore")

    # 创建问答系统
    qa = PDFQA(vectorstore)

    # 问答
    print("\n" + "=" * 50)
    print("📚 PDF 文档问答系统")
    print("=" * 50)

    while True:
        question = input("\n请输入问题（输入 'quit' 退出）: ").strip()
        if question.lower() == "quit":
            break

        print("\n🔍 检索中...")
        answer, sources = qa.ask(question)

        print(f"\n📝 回答: {answer}")
        print(f"\n📚 参考来源 ({len(sources)} 条):")
        for i, doc in enumerate(sources, 1):
            page = doc.metadata.get("page", "?")
            content = doc.page_content[:100].replace("\n", " ")
            print(f"  {i}. 第{page}页: {content}...")

6.2.2 添加引用高亮

"""
带引用标注的 RAG 问答
"""

from pydantic import BaseModel, Field
from typing import List

class SourceWithCitation(BaseModel):
    content: str = Field(description="相关文档内容")
    page: int = Field(description="页码")
    score: float = Field(description="相似度分数")

class AnswerWithCitations(BaseModel):
    answer: str = Field(description="回答内容")
    sources: List[SourceWithCitation] = Field(description="引用来源")

def answer_with_citations(qa: PDFQA, question: str) -> AnswerWithCitations:
    """返回带引用的回答"""
    # 获取答案
    answer, docs = qa.ask(question)

    # 获取带分数的检索结果
    docs_with_scores = qa.vectorstore.similarity_search_with_score(
        question,
        k=3
    )

    sources = [
        SourceWithCitation(
            content=doc.page_content,
            page=doc.metadata.get("page", 1),
            score=score
        )
        for doc, score in docs_with_scores
    ]

    return AnswerWithCitations(
        answer=answer,
        sources=sources
    )

# 使用
result = answer_with_citations(qa, "文档的主要观点是什么？")
print(f"回答: {result.answer}")
print(f"\n引用:")
for i, src in enumerate(result.sources, 1):
    print(f"[{i}] 分数: {src.score:.3f}, 页码: {src.page}")
    print(f"    {src.content[:80]}...")

6.3 项目三：研究代理

构建一个自主研究代理，可以搜索网络、分析信息并生成研究报告。

6.3.1 完整代码

"""
研究代理 - 自动搜索、分析并生成报告
pip install langchain langchain-openai langchain-community duckduckgo-search
"""

import os
from typing import List, Dict, Any
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.tools import tool
from langchain.agents import create_openai_functions_agent, AgentExecutor
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_community.tools import WikipediaQueryRun

# ==================== 工具定义 ====================

# 网络搜索
search = DuckDuckGoSearchRun()

@tool
def search_web(query: str) -> str:
    """搜索网络获取最新信息

    Args:
        query: 搜索关键词
    """
    return search.run(query)

# 维基百科
wikipedia = WikipediaQueryRun()

@tool
def get_wikipedia(topic: str) -> str:
    """从维基百科获取主题的概述信息

    Args:
        topic: 主题名称
    """
    result = wikipedia.run(topic)
    return result

@tool
def save_note(content: str) -> str:
    """保存研究笔记

    Args:
        content: 笔记内容
    """
    with open("research_notes.md", "a", encoding="utf-8") as f:
        f.write(f"\n## {content}\n")
    return "笔记已保存"

# ==================== 初始化 ====================
llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0
)

tools = [search_web, get_wikipedia, save_note]

# ==================== Agent 提示 ====================
RESEARCH_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """你是一个专业的研究助手。帮助用户深入研究各种主题。

你的工作流程：
1. 理解研究目标
2. 使用搜索工具获取相关信息
3. 从多个来源收集信息
4. 保存重要发现到笔记
5. 综合分析后给出结论

注意：
- 搜索时使用多个关键词以获得更全面的信息
- 对于技术性话题，查阅维基百科获取基础概念
- 将重要发现保存到笔记中
- 最后综合所有信息给出完整的研究报告"""),
    MessagesPlaceholder("chat_history", optional=True),
    ("human", "{input}"),
    MessagesPlaceholder("agent_scratchpad"),
])

# ==================== 创建 Agent ====================
agent = create_openai_functions_agent(
    llm=llm,
    tools=tools,
    prompt=RESEARCH_PROMPT
)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=10,  # 最大迭代次数
    handle_parsing_errors=True
)

# ==================== 使用示例 ====================
if __name__ == "__main__":
    print("=" * 60)
    print("🔬 研究代理")
    print("=" * 60)
    print("\n示例问题：")
    print("1. 研究一下人工智能对就业市场的影响")
    print("2. 了解量子计算的最新进展")
    print("3. 调查电动汽车的市场趋势")
    print("\n输入 'quit' 退出\n")

    while True:
        query = input("请输入研究课题: ").strip()
        if query.lower() == "quit":
            print("再见！")
            break

        print("\n🔍 开始研究...\n")

        result = agent_executor.invoke({"input": query})

        print("\n" + "=" * 60)
        print("📊 研究报告")
        print("=" * 60)
        print(result["output"])

        # 显示使用的工具
        if "intermediate_steps" in result:
            print("\n使用的工具:")
            for step in result["intermediate_steps"]:
                tool_name = step[0].tool
                tool_input = str(step[0].tool_input)[:50]
                print(f"  - {tool_name}: {tool_input}...")

6.3.2 生成格式化报告

"""
生成结构化研究报告
"""

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
from typing import List

llm = ChatOpenAI(model="gpt-4o", temperature=0)

class ReportSection(BaseModel):
    title: str = Field(description="章节标题")
    content: str = Field(description="章节内容")
    sources: List[str] = Field(description="信息来源")

class ResearchReport(BaseModel):
    topic: str = Field(description="研究主题")
    summary: str = Field(description="执行摘要")
    sections: List[ReportSection] = Field(description="报告章节")
    conclusion: str = Field(description="结论")
    references: List[str] = Field(description="参考文献")

def generate_report(research_notes: str, topic: str) -> ResearchReport:
    """从研究笔记生成结构化报告"""

    prompt = ChatPromptTemplate.from_template("""基于以下研究笔记，生成一份结构化的研究报告。

研究主题：{topic}

研究笔记：
{notes}

请生成包含以下部分的报告：
1. 执行摘要
2. 多个详细章节
3. 结论
4. 参考文献""")

    structured_llm = llm.with_structured_output(ResearchReport)

    chain = prompt | structured_llm

    return chain.invoke({
        "topic": topic,
        "notes": research_notes
    })

# 使用
with open("research_notes.md", "r", encoding="utf-8") as f:
    notes = f.read()

report = generate_report(notes, "人工智能对就业的影响")

print(f"# {report.topic}")
print(f"\n## 执行摘要\n{report.summary}")
for section in report.sections:
    print(f"\n## {section.title}\n{section.content}")
print(f"\n## 结论\n{report.conclusion}")
print(f"\n## 参考文献\n" + "\n".join(f"- {ref}" for ref in report.references))

6.4 本章小结

本章通过三个实战项目，展示了如何将 LangChain 的知识转化为实际应用：