大模型应用案例¶

from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.chains.history_aware_retriever import create_history_aware_retriever
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI

class IntelligentCustomerService:
    """智能客服系统"""

    def __init__(self, knowledge_base_path: str, api_key: str):
        """
        初始化智能客服系统

        Args:
            knowledge_base_path: 知识库路径
            api_key: OpenAI API密钥
        """
        self.api_key = api_key
        self.store = {}  # 会话历史存储

        # 加载知识库
        self.knowledge_base = self._load_knowledge_base(knowledge_base_path)

        # 构建向量存储
        self.vectorstore = self._build_vectorstore()

        # 初始化对话链
        self.chain = self._build_chain()

    def _load_knowledge_base(self, path: str) -> list:
        """加载知识库"""
        # 这里简化处理,实际应该从文件加载
        return [
            "产品A是一款高性能的AI助手",
            "产品B的价格是99元",
            "退款流程:联系客服→审核→退款"
        ]

    def _build_vectorstore(self):
        """构建向量存储"""
        embeddings = OpenAIEmbeddings(openai_api_key=self.api_key)
        vectorstore = Chroma.from_texts(
            texts=self.knowledge_base,
            embedding=embeddings
        )
        return vectorstore

    def _get_session_history(self, session_id: str) -> BaseChatMessageHistory:
        """获取会话历史"""
        if session_id not in self.store:
            self.store[session_id] = ChatMessageHistory()
        return self.store[session_id]

    def _build_chain(self):
        """构建对话链(LCEL方式)"""
        llm = ChatOpenAI(
            model_name="gpt-4o",
            temperature=0.3,
            openai_api_key=self.api_key
        )

        retriever = self.vectorstore.as_retriever()

        # 1. 创建历史感知检索器(将对话历史融入检索query)
        contextualize_q_prompt = ChatPromptTemplate.from_messages([
            ("system", "根据对话历史和最新问题,生成一个独立的检索查询。"),
            MessagesPlaceholder("chat_history"),
            ("human", "{input}"),
        ])
        history_aware_retriever = create_history_aware_retriever(
            llm, retriever, contextualize_q_prompt
        )

        # 2. 创建问答链
        qa_prompt = ChatPromptTemplate.from_messages([
            ("system", "你是一个专业的智能客服。根据以下检索到的知识回答用户问题。\n\n{context}"),
            MessagesPlaceholder("chat_history"),
            ("human", "{input}"),
        ])
        question_answer_chain = create_stuff_documents_chain(llm, qa_prompt)

        # 3. 组合为带检索的对话链
        rag_chain = create_retrieval_chain(history_aware_retriever, question_answer_chain)

        # 4. 包装为带历史记忆的链
        conversational_chain = RunnableWithMessageHistory(
            rag_chain,
            self._get_session_history,
            input_messages_key="input",
            history_messages_key="chat_history",
            output_messages_key="answer",
        )

        return conversational_chain

    def chat(self, user_input: str, session_id: str = "default") -> str:
        """
        对话

        Args:
            user_input: 用户输入
            session_id: 会话ID,用于区分不同用户会话

        Returns:
            客服回复
        """
        result = self.chain.invoke(
            {"input": user_input},
            config={"configurable": {"session_id": session_id}},
        )
        return result["answer"]

# 使用示例
customer_service = IntelligentCustomerService(
    knowledge_base_path="knowledge_base.txt",
    api_key="your-api-key"
)

# 对话
while True:
    user_input = input("用户: ")
    if user_input.lower() in ['exit', 'quit']:
        break

    response = customer_service.chat(user_input)
    print(f"客服: {response}\n")

from openai import OpenAI

class CodeAssistant:
    """代码生成助手"""

    def __init__(self, api_key: str):
        """
        初始化代码助手

        Args:
            api_key: OpenAI API密钥
        """
        self.client = OpenAI(api_key=api_key)

    def generate_code(self, description: str, language: str = "python") -> dict:
        """
        生成代码

        Args:
            description: 代码描述
            language: 编程语言

        Returns:
            生成的代码和解释
        """
        prompt = f"""
你是一个专业的{language}程序员。请根据以下描述生成代码:

描述: {description}

要求:
1. 生成完整、可运行的代码
2. 添加必要的注释
3. 遵循最佳实践
4. 代码风格清晰

请生成代码:
"""

        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "你是一个专业的程序员。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.3
        )

        code = response.choices[0].message.content

        # 生成代码解释
        explanation = self._explain_code(code, language)

        return {
            'code': code,
            'explanation': explanation,
            'language': language
        }

    def _explain_code(self, code: str, language: str) -> str:
        """解释代码"""
        prompt = f"""
请解释以下{language}代码:

from langchain_community.document_loaders import TextLoader
from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import ChatPromptTemplate

class DocumentQA:
    """文档问答系统"""

    def __init__(self, documents_path: str, api_key: str):
        """
        初始化文档问答系统

        Args:
            documents_path: 文档路径
            api_key: OpenAI API密钥
        """
        self.api_key = api_key

        # 加载文档
        self.documents = self._load_documents(documents_path)

        # 分割文档
        self.splits = self._split_documents()

        # 构建向量存储
        self.vectorstore = self._build_vectorstore()

        # 构建问答链
        self.qa_chain = self._build_qa_chain()

    def _load_documents(self, path: str) -> list:
        """加载文档（支持目录或单个文件）"""
        documents = []
        from pathlib import Path
        p = Path(path)

        if p.is_dir():
            # 遍历目录中的文件，按扩展名选择合适的加载器
            for file in p.rglob("*"):
                if file.suffix == ".pdf":
                    documents.extend(PyPDFLoader(str(file)).load())
                elif file.suffix in (".txt", ".md"):
                    documents.extend(TextLoader(str(file)).load())
        else:
            # 单个文件，按扩展名选择加载器
            if p.suffix == ".pdf":
                documents.extend(PyPDFLoader(path).load())
            else:
                documents.extend(TextLoader(path).load())

        return documents

    def _split_documents(self) -> list:
        """分割文档"""
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200
        )
        splits = text_splitter.split_documents(self.documents)
        return splits

    def _build_vectorstore(self):
        """构建向量存储"""
        embeddings = OpenAIEmbeddings(openai_api_key=self.api_key)
        vectorstore = Chroma.from_documents(
            documents=self.splits,
            embedding=embeddings
        )
        return vectorstore

    def _build_qa_chain(self):
        """构建问答链（LCEL方式，替代已废弃的 RetrievalQA）"""
        llm = ChatOpenAI(
            model_name="gpt-4o",
            temperature=0.3,
            openai_api_key=self.api_key
        )

        system_prompt = """请基于以下上下文回答问题。如果上下文中没有相关信息，请明确说明。

{context}"""
        prompt = ChatPromptTemplate.from_messages([
            ("system", system_prompt),
            ("human", "{input}")
        ])

        retriever = self.vectorstore.as_retriever(search_kwargs={"k": 3})
        question_answer_chain = create_stuff_documents_chain(llm, prompt)
        qa_chain = create_retrieval_chain(retriever, question_answer_chain)

        return qa_chain

    def ask(self, question: str) -> dict:
        """
        提问

        Args:
            question: 问题

        Returns:
            答案和来源
        """
        result = self.qa_chain.invoke({"input": question})

        return {
            'question': question,
            'answer': result['answer'],
            'source_documents': result['context']
        }

# 使用示例
doc_qa = DocumentQA(
    documents_path="./documents/",
    api_key="your-api-key"
)

# 提问
question = "公司的退款政策是什么?"
result = doc_qa.ask(question)

print(f"问题: {result['question']}")
print(f"答案: {result['answer']}")
print("\n来源文档:")
for i, doc in enumerate(result['source_documents'], 1):  # enumerate同时获取索引和元素
    print(f"\n文档{i}:")
    print(doc.page_content[:200] + "...")

def generate_code(description):
    prompt = f"生成代码: {description}"
    # 调用大模型生成代码
    return generated_code