📍 你在这里 AI 应用 · 第 5 篇 / 共 8 篇
← 上一篇 AI 应用开发新范式：Agent 与 MCP · 下一篇 → LangGraph 入门：与流程图的本质区别
LangChain 是一个用于构建基于大型语言模型（LLM）的应用程序的开源框架，它通过提供一系列工具、组件和接口，简化了将 LLM 与外部数据、其他系统进行集成的过程，帮助开发者更高效地构建复杂的 AI 应用。以下从核心概念、主要功能、应用场景等方面进行详细介绍：
一、核心定位与目标
LangChain 的核心目标是解决 LLM 应用开发中的关键挑战：
- LLM 本身的知识局限（基于训练数据，无法实时获取新信息）；
- 与外部数据的交互（如数据库、文件、API 等）；
- 复杂任务的拆解与流程控制（如多步推理、逻辑链构建）。
通过模块化设计，LangChain 让开发者可以像搭积木一样组合不同组件，快速实现从简单问答到复杂智能代理（Agent）的各类应用。
二、核心概念与组件
LangChain 的核心组件围绕“链（Chain）”和“代理（Agent）”展开，主要包括以下模块：
1. 模型（Models）
- LLM 包装器：支持集成主流 LLM（如 GPT-4、Claude、LLaMA 等），统一调用接口，简化不同模型的切换。
- 聊天模型（Chat Models）：针对对话场景优化，支持基于消息列表（而非纯文本）的交互（如区分“用户消息”“系统提示”）。
- 文本嵌入模型（Embedding Models）：将文本转换为向量（用于相似度计算、检索等），支持 OpenAI Embeddings、Hugging Face 模型等。
2. 提示词（Prompts）
- 提示词模板（Prompt Templates）：标准化提示词格式，支持动态插入变量（如用户输入、上下文信息），避免重复编写固定文本。
- 示例：
"请总结以下内容：{user_input}"
- 示例：
- 提示词选择器（Prompt Selectors）：根据条件（如输入长度、模型类型）自动选择合适的提示词模板。
- 输出解析器（Output Parsers）：将 LLM 的原始输出转换为结构化格式（如 JSON、列表），方便后续处理。
3. 链（Chains）
- 链是 LangChain 的核心概念，用于将多个组件（如模型、工具、提示词）按逻辑串联，实现复杂功能。
- 基础链（LLMChain）：最简单的链，将提示词模板与 LLM 结合，直接输出结果。
- 组合链（Combination Chains）：
SequentialChain：按顺序执行多个链，前一个链的输出作为后一个的输入。RouterChain：根据输入内容动态选择不同的子链执行（类似“路由”）。
- 工具链（Tool-related Chains）：结合外部工具（如搜索引擎、数据库）的链，如
RetrievalQAChain（检索增强问答）。
4. 数据连接（Data Connection）
- 解决 LLM 与外部数据的交互问题，核心是“检索增强生成（RAG）”流程：
- 文档加载器（Document Loaders）：加载各种来源的文档（如 PDF、Word、网页、数据库等）。
- 文档分割器（Text Splitters）：将长文档拆分为短片段（适配 LLM 上下文窗口）。
- 向量存储（Vectorstores）：存储文档片段的向量表示（如 Pinecone、Chroma、FAISS 等），用于快速相似度检索。
- 检索器（Retrievers）：从向量存储中检索与问题相关的文档片段，作为 LLM 生成答案的参考。
5. 代理（Agents）
- 代理是具有“决策能力”的组件，能根据用户需求自主选择工具（如搜索、计算、调用 API）并执行多步任务。
- 工具（Tools）：代理可调用的外部功能，如
SerpAPI（搜索引擎）、PythonREPL（执行代码）、SQLDatabaseToolkit（操作数据库）等。 - 代理类型：
ZeroShotAgent：基于单轮提示词决定调用工具。ReActAgent：遵循“思考-行动-观察”循环（Think-Act-Observe），适合复杂推理。StructuredChatAgent：支持结构化输出（如 JSON），适合精确工具调用。
6. 记忆（Memory）
- 让链或代理具备“记忆”能力，保存对话历史或中间结果，支持多轮交互。
- 短期记忆：如
ConversationBufferMemory（保存完整对话历史）、ConversationSummaryMemory（保存对话摘要）。 - 长期记忆：结合外部存储（如数据库）保存长期对话记录。
三、主要功能与优势
- 模块化与可扩展性：所有组件均可独立替换（如换模型、换工具），方便适配不同场景。
- 简化复杂流程：封装了 RAG、Agent 等复杂逻辑，开发者无需从零实现。
- 跨平台兼容性：支持 Python 和 JavaScript/TypeScript 两种主流语言，适配多环境开发。
- 丰富的生态集成：已对接数百种工具、模型和存储服务（如 OpenAI、AWS、Google Cloud 等）。
四、典型应用场景
- 检索增强问答（RAG）：结合企业文档、知识库，让 LLM 生成基于特定数据的答案（如客服系统、内部知识库查询）。
- 智能代理（Agent）：自动完成多步任务（如“查天气→订机票→写行程”“分析数据→生成图表→撰写报告”）。
- 对话机器人：支持多轮对话，具备记忆能力（如客服机器人、聊天助手）。
- 代码生成与分析：结合代码解释器，实现自动写代码、调试或数据分析。
- 文档处理：自动总结、翻译、提取信息（如合同分析、论文摘要生成）。
五、学习与使用资源
- 官方文档：LangChain 文档（含快速入门、API 详解、示例）。
- 社区生态：活跃的 GitHub 仓库（langchain-ai/langchain）、Discord 社区和教程资源。
- 入门示例：通过简单代码即可实现 RAG、对话记忆等功能，门槛较低。
总之，LangChain 是 LLM 应用开发的“瑞士军刀”，它降低了构建复杂 AI 系统的难度，让开发者更专注于业务逻辑而非底层集成，是当前构建企业级 LLM 应用的主流框架之一。
import os
import re
from typing import List, Dict, Any, Optional
from langchain.embeddings import OpenAIEmbeddings, HuggingFaceEmbeddings
from langchain.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.llms import OpenAI, HuggingFaceHub
from langchain.chains import RetrievalQA
from langchain.document_loaders import DirectoryLoader, TextLoader, PyPDFLoader
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationalRetrievalChain
from langchain.prompts import PromptTemplate
from langchain.callbacks import get_openai_callback
import gradio as gr
# 配置类 - 管理所有参数和设置
class Config:
def __init__(self):
# 大语言模型配置
self.llm_provider = "openai" # 可选: openai, huggingface
self.openai_api_key = os.environ.get("OPENAI_API_KEY", "")
self.huggingface_api_token = os.environ.get("HUGGINGFACEHUB_API_TOKEN", "")
self.llm_model = "gpt-3.5-turbo" # OpenAI模型
# self.llm_model = "google/flan-t5-xl" # HuggingFace模型
# 文档处理配置
self.data_directory = "docs" # 存放文档的目录
self.embedding_model = "text-embedding-ada-002" # 嵌入模型
# self.embedding_model = "sentence-transformers/all-MiniLM-L6-v2"
# 向量数据库配置
self.persist_directory = "chroma_db"
# 链配置
self.k = 4 # 检索的文档数量
self.chain_type = "stuff" # 可选: stuff, map_reduce, refine, map_rerank
self.return_source_documents = True
# 界面配置
self.use_gradio = True # 是否使用Gradio界面
# 文档处理类 - 加载和处理本地文档
class DocumentProcessor:
def __init__(self, config: Config):
self.config = config
def load_documents(self) -> List[Any]:
"""加载并解析指定目录中的所有文档"""
print(f"正在从 {self.config.data_directory} 加载文档...")
# 支持多种文件类型的加载器
loaders = [
("**/*.txt", TextLoader),
("**/*.pdf", PyPDFLoader),
# 可以添加更多文件类型支持
]
all_documents = []
for pattern, loader_class in loaders:
try:
loader = DirectoryLoader(
self.config.data_directory,
glob=pattern,
loader_cls=loader_class
)
documents = loader.load()
all_documents.extend(documents)
print(f"已加载 {len(documents)} 个 {pattern[3:]} 文件")
except Exception as e:
print(f"加载 {pattern[3:]} 文件时出错: {e}")
return all_documents
def process_documents(self, documents: List[Any]) -> List[Any]:
"""处理文档：分割文本为较小的块"""
if not documents:
print("没有找到文档，请检查文档目录")
return []
print(f"正在处理 {len(documents)} 个文档...")
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200
)
split_documents = text_splitter.split_documents(documents)
print(f"文档已分割为 {len(split_documents)} 个文本块")
return split_documents
# 向量数据库类 - 管理文档的向量化存储和检索
class VectorDatabase:
def __init__(self, config: Config):
self.config = config
self.embeddings = self._create_embeddings()
self.db = None
def _create_embeddings(self) -> Any:
"""创建嵌入模型实例"""
if self.config.llm_provider == "openai":
return OpenAIEmbeddings(
model=self.config.embedding_model,
openai_api_key=self.config.openai_api_key
)
else:
return HuggingFaceEmbeddings(
model_name=self.config.embedding_model,
model_kwargs={"device": "cpu"} # 使用CPU，如需GPU请改为"cuda"
)
def create_or_load_db(self, documents: List[Any]) -> None:
"""创建新的向量数据库或加载已存在的数据库"""
if documents and len(documents) > 0:
print("正在创建新的向量数据库...")
self.db = Chroma.from_documents(
documents=documents,
embedding=self.embeddings,
persist_directory=self.config.persist_directory
)
self.db.persist()
print(f"向量数据库已保存到 {self.config.persist_directory}")
else:
print("正在加载已存在的向量数据库...")
self.db = Chroma(
embedding_function=self.embeddings,
persist_directory=self.config.persist_directory
)
print("向量数据库加载成功")
def get_retriever(self) -> Any:
"""获取检索器实例"""
if self.db is None:
raise ValueError("向量数据库未初始化")
return self.db.as_retriever(
search_kwargs={"k": self.config.k}
)
# 大语言模型类 - 管理与LLM的交互
class LLMManager:
def __init__(self, config: Config):
self.config = config
self.llm = self._create_llm()
def _create_llm(self) -> Any:
"""创建大语言模型实例"""
if self.config.llm_provider == "openai":
if not self.config.openai_api_key:
raise ValueError("OpenAI API密钥未设置")
return OpenAI(
model_name=self.config.llm_model,
openai_api_key=self.config.openai_api_key,
temperature=0.1 # 较低的温度使输出更确定性
)
else:
if not self.config.huggingface_api_token:
raise ValueError("HuggingFace API令牌未设置")
return HuggingFaceHub(
repo_id=self.config.llm_model,
huggingfacehub_api_token=self.config.huggingface_api_token,
model_kwargs={"temperature": 0.1, "max_length": 512}
)
# 对话链类 - 构建完整的对话检索链
class ConversationChain:
def __init__(self, config: Config, retriever: Any, llm: Any):
self.config = config
self.llm = llm
self.retriever = retriever
self.memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True
)
self.chain = self._create_chain()
def _create_chain(self) -> Any:
"""创建对话检索链"""
# 自定义提示模板，使回答更符合需求
template = """
你是一个智能助手，帮助用户解答问题。请根据以下已知信息回答问题。
如果已知信息不足，请直接说你不知道，不要编造信息。
对话历史: {chat_history}
用户问题: {question}
已知信息: {context}
回答:
"""
QA_CHAIN_PROMPT = PromptTemplate(
input_variables=["chat_history", "question", "context"],
template=template,
)
# 创建对话检索链
return ConversationalRetrievalChain.from_llm(
self.llm,
retriever=self.retriever,
memory=self.memory,
chain_type=self.config.chain_type,
return_source_documents=self.config.return_source_documents,
combine_docs_chain_kwargs={"prompt": QA_CHAIN_PROMPT}
)
def ask(self, question: str) -> Dict[str, Any]:
"""向链提问并获取回答"""
with get_openai_callback() as cb: # 仅在使用OpenAI时有效
result = self.chain({"question": question})
# 处理回答，提取来源
answer = result["answer"].strip()
source_documents = result.get("source_documents", [])
sources = []
for doc in source_documents:
source = os.path.basename(doc.metadata.get("source", ""))
page = doc.metadata.get("page", "N/A")
sources.append(f"{source} (第{page}页)" if page != "N/A" else source)
# 整理回答
response = {
"answer": answer,
"sources": sources if sources else ["无"],
"tokens_used": getattr(cb, "total_tokens", 0) if self.config.llm_provider == "openai" else 0
}
return response
# 界面类 - 提供用户交互界面
class ChatInterface:
def __init__(self, conversation_chain: ConversationChain, config: Config):
self.conversation_chain = conversation_chain
self.config = config
def run_cli(self) -> None:
"""运行命令行界面"""
print("\n===== 智能对话机器人已启动 =====")
print("输入问题开始对话，输入'退出'结束对话")
while True:
user_input = input("\n你: ")
if user_input.lower() in ["退出", "q", "quit", "exit"]:
break
if not user_input.strip():
continue
response = self.conversation_chain.ask(user_input)
print(f"\n助手: {response['answer']}")
if response["sources"] and response["sources"][0] != "无":
print("\n参考资料:")
for i, source in enumerate(response["sources"], 1):
print(f" {i}. {source}")
if response["tokens_used"] > 0:
print(f"\n本次对话使用了 {response['tokens_used']} 个令牌")
def run_gradio(self) -> None:
"""运行Gradio Web界面"""
def chat_interface(user_message, history):
response = self.conversation_chain.ask(user_message)
bot_message = response["answer"]
# 添加参考资料
if response["sources"] and response["sources"][0] != "无":
sources_text = "\n\n参考资料:\n" + "\n".join(
[f"• {source}" for source in response["sources"]]
)
bot_message += sources_text
history.append((user_message, bot_message))
return history, ""
print("\n===== Gradio Web界面已启动 =====")
print("在浏览器中打开 http://localhost:7860 访问对话界面")
with gr.Blocks(title="智能对话机器人") as demo:
chatbot = gr.Chatbot(height=500)
msg = gr.Textbox(label="输入问题", placeholder="输入问题，按Enter发送，输入'退出'结束对话")
clear = gr.Button("清空对话")
msg.submit(chat_interface, [msg, chatbot], [chatbot, msg])
clear.click(lambda: None, None, chatbot, queue=False)
demo.launch()
# 主函数 - 整合所有组件
def main():
# 加载配置
config = Config()
try:
# 初始化文档处理器
doc_processor = DocumentProcessor(config)
documents = doc_processor.load_documents()
processed_docs = doc_processor.process_documents(documents)
# 初始化向量数据库
vector_db = VectorDatabase(config)
vector_db.create_or_load_db(processed_docs)
retriever = vector_db.get_retriever()
# 初始化LLM管理器
llm_manager = LLMManager(config)
llm = llm_manager.llm
# 初始化对话链
conv_chain = ConversationChain(config, retriever, llm)
# 初始化界面
chat_interface = ChatInterface(conv_chain, config)
# 启动界面
if config.use_gradio:
chat_interface.run_gradio()
else:
chat_interface.run_cli()
except Exception as e:
print(f"程序运行出错: {e}")
if __name__ == "__main__":
main()
import os
import re
from typing import List, Dict, Any, Optional
from langchain.embeddings import OpenAIEmbeddings, HuggingFaceEmbeddings, QianWenEmbeddings
from langchain.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.llms import OpenAI, HuggingFaceHub, QwenLMEndpoint
from langchain.chains import RetrievalQA
from langchain.document_loaders import DirectoryLoader, TextLoader, PyPDFLoader
from langchain.memory import ConversationBufferWindowMemory
from langchain.chains import ConversationalRetrievalChain
from langchain.prompts import PromptTemplate
from langchain.callbacks import get_openai_callback
import gradio as gr
class Config:
"""配置类 - 管理所有参数和设置"""
def __init__(self):
# 大语言模型配置
self.llm_provider = "qwen" # 可选: openai, huggingface, qwen
self.openai_api_key = os.environ.get("OPENAI_API_KEY", "")
self.huggingface_api_token = os.environ.get("HUGGINGFACEHUB_API_TOKEN", "")
self.qwen_api_key = os.environ.get("QWEN_API_KEY", "")
self.llm_model = "qwen-turbo" # Qwen模型
# 文档处理配置
self.data_directory = "docs" # 存放文档的目录
self.embedding_model = "text-embedding-ada-002" # 嵌入模型
# 向量数据库配置
self.persist_directory = "chroma_db"
# 链配置
self.k = 4 # 检索的文档数量
self.chain_type = "stuff" # 可选: stuff, map_reduce, refine, map_rerank
self.return_source_documents = True
# 界面配置
self.use_gradio = True # 是否使用Gradio界面
# 记忆配置
self.memory_window_size = 3 # 保存的对话轮数
class DocumentProcessor:
"""文档处理类 - 加载和处理本地文档"""
def __init__(self, config: Config):
self.config = config
def load_documents(self) -> List[Any]:
"""
加载并解析指定目录中的所有文档
支持多种文件类型，包括文本和PDF
"""
print(f"正在从 {self.config.data_directory} 加载文档...")
# 定义支持的文件类型及其对应的加载器
loaders = [
("**/*.txt", TextLoader),
("**/*.pdf", PyPDFLoader),
# 可以添加更多文件类型支持
]
all_documents = []
for pattern, loader_class in loaders:
try:
loader = DirectoryLoader(
self.config.data_directory,
glob=pattern,
loader_cls=loader_class
)
documents = loader.load()
all_documents.extend(documents)
print(f"已加载 {len(documents)} 个 {pattern[3:]} 文件")
except Exception as e:
print(f"加载 {pattern[3:]} 文件时出错: {e}")
return all_documents
def process_documents(self, documents: List[Any]) -> List[Any]:
"""
处理文档：将文本分割为较小的块
便于后续向量化和检索
"""
if not documents:
print("没有找到文档，请检查文档目录")
return []
print(f"正在处理 {len(documents)} 个文档...")
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000, # 每个文本块的最大长度
chunk_overlap=200 # 文本块之间的重叠长度，确保上下文连贯
)
split_documents = text_splitter.split_documents(documents)
print(f"文档已分割为 {len(split_documents)} 个文本块")
return split_documents
class VectorDatabase:
"""向量数据库类 - 管理文档的向量化存储和检索"""
def __init__(self, config: Config):
self.config = config
self.embeddings = self._create_embeddings()
self.db = None
def _create_embeddings(self) -> Any:
"""
创建嵌入模型实例
根据配置选择不同的嵌入模型
"""
if self.config.llm_provider == "openai":
return OpenAIEmbeddings(
model=self.config.embedding_model,
openai_api_key=self.config.openai_api_key
)
elif self.config.llm_provider == "qwen" and self.config.embedding_model == "qwen-embedding":
return QianWenEmbeddings(
qwen_api_key=self.config.qwen_api_key
)
else:
return HuggingFaceEmbeddings(
model_name=self.config.embedding_model,
model_kwargs={"device": "cpu"} # 使用CPU，如需GPU请改为"cuda"
)
def create_or_load_db(self, documents: List[Any]) -> None:
"""
创建新的向量数据库或加载已存在的数据库
如果提供了文档，则创建新数据库；否则加载现有数据库
"""
if documents and len(documents) > 0:
print("正在创建新的向量数据库...")
self.db = Chroma.from_documents(
documents=documents,
embedding=self.embeddings,
persist_directory=self.config.persist_directory
)
self.db.persist()
print(f"向量数据库已保存到 {self.config.persist_directory}")
else:
print("正在加载已存在的向量数据库...")
self.db = Chroma(
embedding_function=self.embeddings,
persist_directory=self.config.persist_directory
)
print("向量数据库加载成功")
def get_retriever(self) -> Any:
"""获取检索器实例，用于从向量数据库中检索相关文档"""
if self.db is None:
raise ValueError("向量数据库未初始化")
return self.db.as_retriever(
search_kwargs={"k": self.config.k} # 返回最相关的k个文档
)
class LLMManager:
"""大语言模型类 - 管理与LLM的交互"""
def __init__(self, config: Config):
self.config = config
self.llm = self._create_llm()
def _create_llm(self) -> Any:
"""
创建大语言模型实例
根据配置选择不同的LLM提供商和模型
"""
if self.config.llm_provider == "openai":
if not self.config.openai_api_key:
raise ValueError("OpenAI API密钥未设置")
return OpenAI(
model_name=self.config.llm_model,
openai_api_key=self.config.openai_api_key,
temperature=0.1 # 较低的温度使输出更确定性
)
elif self.config.llm_provider == "huggingface":
if not self.config.huggingface_api_token:
raise ValueError("HuggingFace API令牌未设置")
return HuggingFaceHub(
repo_id=self.config.llm_model,
huggingfacehub_api_token=self.config.huggingface_api_token,
model_kwargs={"temperature": 0.1, "max_length": 512}
)
elif self.config.llm_provider == "qwen":
if not self.config.qwen_api_key:
raise ValueError("Qwen API密钥未设置")
return QwenLMEndpoint(
model_name=self.config.llm_model,
qwen_api_key=self.config.qwen_api_key,
temperature=0.1
)
else:
raise ValueError(f"不支持的LLM提供商: {self.config.llm_provider}")
class ConversationChain:
"""对话链类 - 构建完整的对话检索链"""
def __init__(self, config: Config, retriever: Any, llm: Any):
self.config = config
self.llm = llm
self.retriever = retriever
# 使用窗口内存，保留最近的k轮对话
self.memory = ConversationBufferWindowMemory(
memory_key="chat_history",
return_messages=True,
k=config.memory_window_size # 只保留最近的k轮对话
)
self.chain = self._create_chain()
def _create_chain(self) -> Any:
"""创建对话检索链，整合LLM、检索器和记忆"""
# 自定义提示模板，指导模型如何利用上下文和历史回答问题
template = """
你是一个智能助手，帮助用户解答问题。请根据以下已知信息和对话历史回答问题。
如果已知信息不足，请直接说你不知道，不要编造信息。
对话历史: {chat_history}
用户问题: {question}
已知信息: {context}
回答:
"""
QA_CHAIN_PROMPT = PromptTemplate(
input_variables=["chat_history", "question", "context"],
template=template,
)
return ConversationalRetrievalChain.from_llm(
self.llm,
retriever=self.retriever,
memory=self.memory,
chain_type=self.config.chain_type,
return_source_documents=self.config.return_source_documents,
combine_docs_chain_kwargs={"prompt": QA_CHAIN_PROMPT}
)
def ask(self, question: str) -> Dict[str, Any]:
"""
向链提问并获取回答
返回包含回答、参考资料和令牌使用量的字典
"""
# 针对OpenAI模型，使用回调统计令牌使用量
if self.config.llm_provider == "openai":
with get_openai_callback() as cb:
result = self.chain({"question": question})
tokens_used = cb.total_tokens
else:
result = self.chain({"question": question})
tokens_used = 0
answer = result["answer"].strip()
source_documents = result.get("source_documents", [])
sources = []
# 提取参考资料信息
for doc in source_documents:
source = os.path.basename(doc.metadata.get("source", ""))
page = doc.metadata.get("page", "N/A")
sources.append(f"{source} (第{page}页)" if page != "N/A" else source)
return {
"answer": answer,
"sources": sources if sources else ["无"],
"tokens_used": tokens_used
}
class ChatInterface:
"""界面类 - 提供用户交互界面"""
def __init__(self, conversation_chain: ConversationChain, config: Config):
self.conversation_chain = conversation_chain
self.config = config
def run_cli(self) -> None:
"""运行命令行界面"""
print("\n===== 智能对话机器人已启动 =====")
print(f"对话记忆: 系统将保留最近的 {self.config.memory_window_size} 轮对话")
print("输入问题开始对话，输入'退出'结束对话")
while True:
user_input = input("\n你: ")
if user_input.lower() in ["退出", "q", "quit", "exit"]:
break
if not user_input.strip():
continue
response = self.conversation_chain.ask(user_input)
print(f"\n助手: {response['answer']}")
if response["sources"] and response["sources"][0] != "无":
print("\n参考资料:")
for i, source in enumerate(response["sources"], 1):
print(f" {i}. {source}")
if response["tokens_used"] > 0:
print(f"\n本次对话使用了 {response['tokens_used']} 个令牌")
def run_gradio(self) -> None:
"""运行Gradio Web界面"""
def chat_interface(user_message, history):
"""Gradio聊天接口函数"""
response = self.conversation_chain.ask(user_message)
bot_message = response["answer"]
# 添加参考资料信息到回答中
if response["sources"] and response["sources"][0] != "无":
sources_text = "\n\n参考资料:\n" + "\n".join(
[f"• {source}" for source in response["sources"]]
)
bot_message += sources_text
history.append((user_message, bot_message))
return history, ""
print("\n===== Gradio Web界面已启动 =====")
print(f"对话记忆: 系统将保留最近的 {self.config.memory_window_size} 轮对话")
print("在浏览器中打开 http://localhost:7860 访问对话界面")
# 构建Gradio界面
with gr.Blocks(title="智能对话机器人") as demo:
gr.Markdown(f"### 智能对话机器人\n对话记忆: 系统将保留最近的 {self.config.memory_window_size} 轮对话")
chatbot = gr.Chatbot(height=500) # 聊天界面
msg = gr.Textbox(label="输入问题", placeholder="输入问题，按Enter发送，输入'退出'结束对话")
clear = gr.Button("清空对话") # 清空对话按钮
# 设置交互逻辑
msg.submit(chat_interface, [msg, chatbot], [chatbot, msg])
clear.click(lambda: None, None, chatbot, queue=False)
demo.launch()
def main():
"""主函数 - 整合所有组件，启动应用"""
# 加载配置
config = Config()
try:
# 初始化文档处理器
doc_processor = DocumentProcessor(config)
documents = doc_processor.load_documents()
processed_docs = doc_processor.process_documents(documents)
# 初始化向量数据库
vector_db = VectorDatabase(config)
vector_db.create_or_load_db(processed_docs)
retriever = vector_db.get_retriever()
# 初始化LLM管理器
llm_manager = LLMManager(config)
llm = llm_manager.llm
# 初始化对话链
conv_chain = ConversationChain(config, retriever, llm)
# 初始化界面
chat_interface = ChatInterface(conv_chain, config)
# 启动界面
if config.use_gradio:
chat_interface.run_gradio()
else:
chat_interface.run_cli()
except Exception as e:
print(f"程序运行出错: {e}")
if __name__ == "__main__":
main()
— 感谢阅读 —
一起交流
分享你的思考，让讨论更进一步。