首页/资讯中心/详情

基于RAG的问答智能体实战案例

基于RAG的问答智能体实战案例
📅 发布时间：2026/6/19 16:38:33

《AI Agent智能体开发实践+玩转FastGPT 像搭积木一样构建智能体 LLM大语言模型AI Agent开发 智能体性能优化调试部署实施方法书籍 AIAgent智能体开发实践 无规格》【摘要 书评 试读】- 京东图书

本章将基于LangChain框架和检索增强生成（RAG）技术实现一个问答智能体，这个智能体能够根据RAG知识准确回答用户的问题。该项目支持单轮问答和多轮对话两种模式，采用RAG技术，通过向量数据库实现对企业FAQ知识的高效检索，并结合大语言模型生成准确回答。

10.1 系统架构

该项目基于LangChain框架和阿里云DashScope服务，构建了一个支持单轮问答与多轮对话的企业FAQ智能问答系统，整体框架可分为6个核心层级，形成“输入−处理−输出”的完整闭环，如图10.1所示。

图10.1 整体框架可分为6个核心层级

1. 数据源层

2. 数据存储层

3. 模型层

4. 链管理层

1）单轮问答链

2）多轮对话链

5. 业务逻辑层

1）单轮问答函数（single_question）

2）多轮对话函数（chat_conversation）

6. 用户交互层

10.2 核心功能模块设计

1. 基础配置与依赖处理

#从正确安装的pydantic中导入BaseModel

from pydantic import BaseModel

#全局配置允许任意类型

class CustomBaseModel(BaseModel):

class Config:

arbitrary_types_allowed = True

#替换LangChain冲突模块

import langchain.memory.summary

langchain.memory.summary.BaseModel = CustomBaseModel

#其他模块替换

上面代码主要解决LangChain与Pydantic之间的兼容性问题，通过自定义BaseModel并替换 LangChain相关模块中的基础模型类，避免因类型检查严格导致的冲突。

2. 阿里云DashScope API配置

DASHSCOPE_API_KEY = "your-dashscope-api-key-here" #替换为实际API密钥

QWEN_VL_ENDPOINT = "https://dashscope.aliyuncs.com/api/v1/services/aigc/ multimodal-generation/generation"

上面代码配置阿里云DashScope服务的API密钥和端点，这是调用阿里云大模型服务的基础。

3. 自定义嵌入模型

class DashScopeEmbeddings(Embeddings):

def__init__(self, model_name: str = "text-embedding-v2"):

self.model_name = model_name

def embed_documents(self, texts: List[str]) -> List[List[float]]:

#实现文档嵌入逻辑

...

def embed_query(self, text: str) -> List[float]:

#实现查询嵌入逻辑

...

上面代码实现了基于阿里云DashScope的文本嵌入模型，将文本转换为向量表示，用于后续的相似度检索。其包含以下两个核心方法。

4. 自定义LLM模型

class DashScopeLLM(LLM, CustomBaseModel):

model_name: str = "qwen-plus"

temperature: float = 0.3

max_tokens: int = 512

@property

def_llm_type(self) -> str:

return "dashscope"

def_call(self, prompt: str, stop: Optional[List[str]] = None, **kwargs: Any) -> str:

#实现调用阿里云大模型的逻辑

...

上面代码封装了阿里云DashScope的大语言模型调用，实现了LangChain的LLM接口，使得阿里云的模型可以无缝集成到LangChain的工作流中。

5. 系统初始化组件

def initialize_system():

#环境配置、数据准备、组件初始化

...

这是系统的核心初始化函数，主要完成：

6. 交互功能实现

7. 前端界面

with gr.Blocks(title="RAG问答系统") as demo:

#界面组件定义与布局

...

使用Gradio创建了友好的Web交互界面，包含两个标签页：

10.4 完整代码及运行结果

本章到了这里，我们实现了基于LangChain开发问答智能体的完整功能。通过LangChain框架，我们可以快速构建基于特定知识库的问答系统，并根据需要进行优化和扩展。

【示例10.1】基于LangChain的完整示例代码main.py，实现一个RAG问答智能体，支持FAQ数据、向量化、检索、生成回答，并包含多轮对话记忆和基础部署建议（LangChain+Gradio+Qwen）。 # -*- coding: utf-8 -*- import os import dashscope import gradio as gr # 从正确安装的pydantic中导入BaseModel from pydantic import BaseModel # 全局配置允许任意类型 class CustomBaseModel(BaseModel): class Config: arbitrary_types_allowed = True # 替换LangChain冲突模块 import langchain.memory.summary langchain.memory.summary.BaseModel = CustomBaseModel import langchain.chains.combine_documents.base langchain.chains.combine_documents.base.BaseModel = CustomBaseModel import langchain_core.language_models.base langchain_core.language_models.base.BaseModel = CustomBaseModel import langchain.prompts.prompt langchain.prompts.prompt.BaseModel = CustomBaseModel # 导入LangChain组件（已包含社区版依赖） from langchain.vectorstores import FAISS 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.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain.schema import Document, AIMessage, HumanMessage from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.memory import SimpleMemory from langchain.llms.base import LLM from typing import Optional, List, Mapping, Any from langchain.embeddings.base import Embeddings # ----------------------------- # 1. 阿里云 DashScope API 配置 # ----------------------------- DASHSCOPE_API_KEY = "your-dashscope-api-key-here" # 替换为实际API密钥 QWEN_VL_ENDPOINT = "https://dashscope.aliyuncs.com/api/v1/services/aigc/ multimodal-generation/generation" # -------------------------- # 2. 自定义阿里云DashScope嵌入模型 # -------------------------- class DashScopeEmbeddings(Embeddings): def __init__(self, model_name: str = "text-embedding-v2"): self.model_name = model_name def embed_documents(self, texts: List[str]) -> List[List[float]]: dashscope.api_key = DASHSCOPE_API_KEY response = dashscope.TextEmbedding.call( model=self.model_name, input=texts ) if hasattr(response, 'output') and 'embeddings' in response.output: return [item["embedding"] for item in response.output["embeddings"]] else: raise Exception(f"嵌入模型调用失败: {getattr(response, 'message', '未知错误')}") def embed_query(self, text: str) -> List[float]: return self.embed_documents([text])[0] # -------------------------- # 3. 自定义阿里云DashScope LLM # -------------------------- class DashScopeLLM(LLM, CustomBaseModel): model_name: str = "qwen-plus" temperature: float = 0.3 max_tokens: int = 512 @property def _llm_type(self) -> str: return "dashscope" def _call( self, prompt: str, stop: Optional[List[str]] = None,** kwargs: Any, ) -> str: if stop is not None: raise ValueError("不支持stop参数") dashscope.api_key = DASHSCOPE_API_KEY response = dashscope.Generation.call( model=self.model_name, prompt=prompt, temperature=self.temperature, max_tokens=self.max_tokens, endpoint=QWEN_VL_ENDPOINT, **kwargs ) if hasattr(response, 'output') and hasattr(response.output, 'text'): return response.output.text else: raise Exception(f"LLM调用失败: {getattr(response, 'message', '未知错误')}") @property def _identifying_params(self) -> Mapping[str, Any]: return { "model_name": self.model_name, "temperature": self.temperature, "max_tokens": self.max_tokens, "endpoint": QWEN_VL_ENDPOINT } # -------------------------- # 4. 初始化系统组件 # -------------------------- def initialize_system(): os.environ["DASHSCOPE_API_KEY"] = DASHSCOPE_API_KEY # 准备FAQ数据 faq_data = [ {"question": "公司的工作时间是什么？", "answer": "工作日为周一至周五，上午9:00到下午6:00。"}, {"question": "如何申请年假？", "answer": "通过HR系统提交休假申请，主管审批后生效。"}, {"question": "有没有远程办公政策？", "answer": "支持混合办公模式，每周可在家工作最多两天。"}, {"question": "加班有补贴吗？", "answer": "是的，超过晚上8点的加班可申请调休或加班费。"} ] # 转换为Document格式 documents = [] for item in faq_data: content = f"问题: {item['question']}\n答案: {item['answer']}" documents.append(Document(page_content=content, metadata={"source": "faq"})) # 文本分割 text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50) split_docs = text_splitter.split_documents(documents) # 初始化嵌入模型和向量库 embeddings = DashScopeEmbeddings(model_name="text-embedding-v2") db = FAISS.from_documents(split_docs, embeddings) retriever = db.as_retriever(search_kwargs={"k": 2}) # 初始化LLM llm = DashScopeLLM( model_name="qwen-plus", temperature=0.3, max_tokens=512 ) # 定义单轮问答提示模板 qa_prompt = ChatPromptTemplate.from_template("""严格使用以下上下文回答问题，不要编造信息。如果上下文没有相关内容，直接说“不知道”。 上下文: {context} 问题: {input} 回答:""") # 创建单轮问答链 combine_docs_chain = create_stuff_documents_chain(llm, qa_prompt) qa_chain = create_retrieval_chain(retriever, combine_docs_chain) # 定义多轮对话提示模板 contextualize_q_prompt = ChatPromptTemplate.from_messages([ MessagesPlaceholder(variable_name="chat_history"), ("human", "{input}"), ("human", "基于上述对话，将问题重述为一个独立的问题，不需要回答。") ]) # 创建历史感知的检索器 history_aware_retriever = create_history_aware_retriever( llm, retriever, contextualize_q_prompt ) # 创建多轮对话链 chat_chain = create_retrieval_chain(history_aware_retriever, combine_docs_chain) # 初始化多轮对话记忆 memory = SimpleMemory(memories={}) return qa_chain, chat_chain, memory # -------------------------- # 5. 定义交互函数 # -------------------------- def single_question(question): if DASHSCOPE_API_KEY == "your-dashscope-api-key-here": return "请先替换代码中的阿里云DashScope API密钥", "" try: qa_chain, _, _ = initialize_system() result = qa_chain.invoke({"input": question}) # 提取参考来源 sources = [] if result.get("context"): sources = [f"[{i+1}] {doc.page_content.split('答案: ')[-1]}" for i, doc in enumerate(result["context"])] return result["answer"], "\n".join(sources) except Exception as e: return f"错误: {str(e)}", "" def chat_conversation(question, chat_history, memory): if DASHSCOPE_API_KEY == "your-dashscope-api-key-here": chat_history.append((question, "请先替换代码中的阿里云DashScope API密钥")) return chat_history, "", memory try: _, chat_chain, memory = initialize_system() # 准备对话历史 messages = [] for human_msg, ai_msg in chat_history: messages.append(HumanMessage(content=human_msg)) messages.append(AIMessage(content=ai_msg)) # 调用对话链 result = chat_chain.invoke({ "input": question, "chat_history": messages }) # 更新对话历史 chat_history.append((question, result["answer"])) return chat_history, "", memory except Exception as e: chat_history.append((question, f"错误: {str(e)}")) return chat_history, "", memory # -------------------------- # 6. 创建Gradio界面 # -------------------------- with gr.Blocks(title="RAG问答系统") as demo: gr.Markdown("# 企业FAQ智能问答系统") gr.Markdown("基于LangChain和阿里云DashScope的RAG问答演示") memory_state = gr.State(None) # 用于存储对话记忆 with gr.Tabs(): with gr.Tab("单轮问答"): question_single = gr.Textbox(label="请输入问题") output_single = gr.Textbox(label="回答") sources_single = gr.Textbox(label="参考来源") btn_single = gr.Button("提交") btn_single.click( single_question, inputs=[question_single], outputs=[output_single, sources_single] ) with gr.Tab("多轮对话"): chatbot = gr.Chatbot(label="对话历史") question_chat = gr.Textbox(label="请输入问题") btn_chat = gr.Button("发送") btn_clear = gr.Button("清空对话") btn_chat.click( chat_conversation, inputs=[question_chat, chatbot, memory_state], outputs=[chatbot, question_chat, memory_state] ) btn_clear.click(lambda: (None, "", None), None, [chatbot, question_chat, memory_state]) question_chat.submit( chat_conversation, inputs=[question_chat, chatbot, memory_state], outputs=[chatbot, question_chat, memory_state] ) gr.Markdown(""" ### 使用说明 1. 请先在代码中替换您的阿里云DashScope API密钥 2. 单轮问答：适合独立的问题查询 3. 多轮对话：支持上下文关联的连续对话 """) if __name__ == "__main__": demo.launch()

运行代码，控制台窗口中输出如下（请读者自行测试）：

Running on local URL: http://127.0.0.1:7862