从零搭建DeepSeek+智能体&知识库：你的专属AI助手实战指南

# 访问DeepSeek官网注册账号# 进入控制台创建API Key# 保存好你的API密钥，后续配置需要

# 创建项目目录mkdir deepseek-agentcd deepseek-agent# 创建虚拟环境python -m venv venv# Windows系统执行: venv\Scripts\activatesource venv/bin/activate  # Mac/Linux# 安装核心依赖pip install openai  # DeepSeek兼容OpenAI接口格式pip install chromadb  # 向量数据库pip install langchain  # 智能体框架pip install gradio  # 快速构建界面pip install pypdf docx2txt  # 文档解析

# config.pyimport os# 设置API密钥os.environ["DEEPSEEK_API_KEY"] = "your-api-key-here"# DeepSeek API配置DEEPSEEK_API_BASE = "https://api.deepseek.com/v1"DEEPSEEK_MODEL = "deepseek-chat"  # 对话模型DEEPSEEK_EMBEDDING_MODEL = "deepseek-embedding"  # 嵌入模型# 向量数据库配置CHROMA_PERSIST_DIR = "./chroma_db"COLLECTION_NAME = "knowledge_base"

# knowledge_base/loader.pyimport osfrom typing import Listfrom langchain.document_loaders import (
    PyPDFLoader,
    TextLoader,
    Docx2txtLoader,
    DirectoryLoader)class DocumentLoader:
    """文档加载器，支持多种格式"""
    
    def __init__(self, docs_path: str):
        self.docs_path = docs_path        
    def load_all_documents(self):
        """加载目录下的所有文档"""
        documents = []
        
        # PDF加载器
        pdf_loader = DirectoryLoader(
            self.docs_path,
            glob="**/*.pdf",
            loader_cls=PyPDFLoader        )
        documents.extend(pdf_loader.load())
        
        # TXT加载器
        txt_loader = DirectoryLoader(
            self.docs_path,
            glob="**/*.txt",
            loader_cls=TextLoader        )
        documents.extend(txt_loader.load())
        
        # Word文档加载器
        docx_loader = DirectoryLoader(
            self.docs_path,
            glob="**/*.docx",
            loader_cls=Docx2txtLoader        )
        documents.extend(docx_loader.load())
        
        print(f"成功加载 {len(documents)} 个文档")
        return documents

# knowledge_base/splitter.pyfrom langchain.text_splitter import RecursiveCharacterTextSplitterclass DocumentSplitter:
    """智能文档分割器"""
    
    def __init__(self, chunk_size=500, chunk_overlap=50):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            length_function=len,
            separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""]
        )
    
    def split_documents(self, documents):
        """分割文档"""
        chunks = self.text_splitter.split_documents(documents)
        print(f"文档分割为 {len(chunks)} 个文本块")
        
        # 为每个块添加元数据
        for i, chunk in enumerate(chunks):
            if not hasattr(chunk, 'metadata'):
                chunk.metadata = {}
            chunk.metadata['chunk_id'] = i
            chunk.metadata['source_file'] = chunk.metadata.get('source', 'unknown')
            
        return chunks

# knowledge_base/vector_store.pyimport chromadbfrom chromadb.config import Settingsfrom langchain.embeddings import OpenAIEmbeddings  # DeepSeek嵌入兼容from langchain.vectorstores import Chromaclass VectorStore:
    """向量数据库管理"""
    
    def __init__(self, persist_directory, collection_name):
        self.persist_directory = persist_directory
        self.collection_name = collection_name        
        # 初始化DeepSeek嵌入模型
        self.embeddings = OpenAIEmbeddings(
            model="deepseek-embedding",
            openai_api_key=os.getenv("DEEPSEEK_API_KEY"),
            openai_api_base="https://api.deepseek.com/v1"
        )
        
        # 初始化Chroma客户端
        self.client = chromadb.PersistentClient(
            path=persist_directory,
            settings=Settings(anonymized_telemetry=False)
        )
    
    def create_vector_store(self, chunks):
        """从文档块创建向量库"""
        vector_store = Chroma.from_documents(
            documents=chunks,
            embedding=self.embeddings,
            persist_directory=self.persist_directory,
            collection_name=self.collection_name        )
        vector_store.persist()
        print(f"向量库创建完成，保存在 {self.persist_directory}")
        return vector_store    
    def load_vector_store(self):
        """加载已有向量库"""
        vector_store = Chroma(
            persist_directory=self.persist_directory,
            embedding_function=self.embeddings,
            collection_name=self.collection_name        )
        return vector_store

# build_knowledge_base.pyimport osfrom knowledge_base.loader import DocumentLoaderfrom knowledge_base.splitter import DocumentSplitterfrom knowledge_base.vector_store import VectorStoredef build_knowledge_base(docs_path="./docs"):
    """一键构建知识库"""
    
    print("开始构建知识库...")
    
    # 1. 加载文档
    loader = DocumentLoader(docs_path)
    documents = loader.load_all_documents()
    
    if not documents:
        print("未找到文档，请检查路径")
        return
    
    # 2. 分割文档
    splitter = DocumentSplitter(chunk_size=500, chunk_overlap=50)
    chunks = splitter.split_documents(documents)
    
    # 3. 向量化存储
    vector_store = VectorStore(
        persist_directory="./chroma_db",
        collection_name="knowledge_base"
    )
    vector_store.create_vector_store(chunks)
    
    print("知识库构建完成！")if __name__ == "__main__":
    build_knowledge_base()

# agent/agent_core.pyfrom langchain.agents import AgentExecutor, create_openai_tools_agentfrom langchain.prompts import ChatPromptTemplate, MessagesPlaceholderfrom langchain_openai import ChatOpenAIfrom langchain.memory import ConversationBufferMemoryclass DeepSeekAgent:
    """DeepSeek智能体核心"""
    
    def __init__(self, tools=None, system_prompt=None):
        # 初始化DeepSeek模型
        self.llm = ChatOpenAI(
            model="deepseek-chat",
            openai_api_key=os.getenv("DEEPSEEK_API_KEY"),
            openai_api_base="https://api.deepseek.com/v1",
            temperature=0.7,
            max_tokens=2000
        )
        
        # 默认系统提示词
        self.system_prompt = system_prompt or """你是一个智能助手，可以回答问题并调用工具完成任务。
        如果遇到知识库相关问题，请使用知识库检索工具获取信息。
        你的回答应当准确、简洁、有帮助。"""        
        # 初始化工具列表
        self.tools = tools or []
        
        # 创建提示模板
        self.prompt = ChatPromptTemplate.from_messages([
            ("system", self.system_prompt),
            MessagesPlaceholder(variable_name="chat_history"),
            ("human", "{input}"),
            MessagesPlaceholder(variable_name="agent_scratchpad")
        ])
        
        # 创建记忆组件
        self.memory = ConversationBufferMemory(
            memory_key="chat_history",
            return_messages=True
        )
        
        # 创建智能体
        self.agent = create_openai_tools_agent(
            llm=self.llm,
            tools=self.tools,
            prompt=self.prompt        )
        
        # 创建执行器
        self.agent_executor = AgentExecutor(
            agent=self.agent,
            tools=self.tools,
            memory=self.memory,
            verbose=True,
            handle_parsing_errors=True
        )
    
    def chat(self, message):
        """与智能体对话"""
        response = self.agent_executor.invoke({"input": message})
        return response["output"]

# tools/retrieval_tool.pyfrom langchain.tools import BaseToolfrom typing import Optional, Typefrom pydantic import BaseModel, Fieldclass KnowledgeBaseInput(BaseModel):
    """知识库检索工具的输入"""
    query: str = Field(description="要检索的问题或关键词")class KnowledgeBaseTool(BaseTool):
    """知识库检索工具"""
    
    name = "knowledge_base_search"
    description = "从知识库中检索相关信息，当用户问题涉及文档内容时使用"
    args_schema: Type[BaseModel] = KnowledgeBaseInput    
    def __init__(self, vector_store):
        super().__init__()
        self.vector_store = vector_store
        self.retriever = vector_store.as_retriever(
            search_type="similarity",
            search_kwargs={"k": 5}
        )
    
    def _run(self, query: str) -> str:
        """执行检索"""
        try:
            docs = self.retriever.get_relevant_documents(query)
            if not docs:
                return "知识库中未找到相关信息"
            
            # 格式化检索结果
            results = []
            for i, doc in enumerate(docs):
                source = doc.metadata.get('source_file', '未知来源')
                content = doc.page_content
                results.append(f"[{i+1}] 来源：{source}\n内容：{content}\n")
            
            return "\n".join(results)
        except Exception as e:
            return f"检索过程中出现错误：{str(e)}"
    
    async def _arun(self, query: str) -> str:
        """异步执行（可选实现）"""
        return self._run(query)

# tools/utility_tools.pyfrom langchain.tools import BaseToolimport requestsimport datetimeimport jsonclass CurrentTimeTool(BaseTool):
    """获取当前时间"""
    name = "get_current_time"
    description = "获取当前日期和时间，当用户询问时间相关问题时使用"
    
    def _run(self, _: str) -> str:
        now = datetime.datetime.now()
        return f"当前时间是：{now.strftime('%Y年%m月%d日 %H:%M:%S')}"class CalculatorTool(BaseTool):
    """简单计算器"""
    name = "calculator"
    description = "执行数学计算，输入应为数学表达式，如 '2+3*4'"
    
    def _run(self, expression: str) -> str:
        try:
            # 安全评估数学表达式
            result = eval(expression, {"__builtins__": {}}, {})
            return f"计算结果：{result}"
        except Exception as e:
            return f"计算错误：{str(e)}"class WebSearchTool(BaseTool):
    """网络搜索（需要配置搜索引擎API）"""
    name = "web_search"
    description = "搜索网络信息，当问题需要实时或最新信息时使用"
    
    def _run(self, query: str) -> str:
        # 这里需要集成搜索引擎API，如必应搜索、百度搜索等
        # 示例返回，实际使用时需要调用真实API
        return f"正在搜索：{query}，请配置搜索引擎API"

# create_agent.pyfrom agent.agent_core import DeepSeekAgentfrom tools.retrieval_tool import KnowledgeBaseToolfrom tools.utility_tools import CurrentTimeTool, CalculatorToolfrom knowledge_base.vector_store import VectorStoredef create_agent_with_knowledge():
    """创建带知识库的智能体"""
    
    # 1. 加载知识库
    vector_store = VectorStore(
        persist_directory="./chroma_db",
        collection_name="knowledge_base"
    ).load_vector_store()
    
    # 2. 创建工具
    tools = [
        KnowledgeBaseTool(vector_store),
        CurrentTimeTool(),
        CalculatorTool()
    ]
    
    # 3. 自定义系统提示词
    system_prompt = """你是一个智能助手，拥有知识库检索能力。
    使用规则：
- 优先使用知识库检索工具回答专业问题
- 对于简单常识问题可以直接回答
- 如果需要计算或查询时间，调用相应工具
- 回答要简洁准确，引用信息来源
- 不确定的信息要明确说明"""    
    # 4. 创建智能体
    agent = DeepSeekAgent(
        tools=tools,
        system_prompt=system_prompt    )
    
    return agentif __name__ == "__main__":
    agent = create_agent_with_knowledge()
    
    # 测试对话
    while True:
        user_input = input("\n请输入问题（输入exit退出）：")
        if user_input.lower() == 'exit':
            break
        
        response = agent.chat(user_input)
        print(f"\n智能体：{response}")

# app.pyimport gradio as grfrom create_agent import create_agent_with_knowledge# 初始化智能体（启动时加载）print("正在初始化智能体...")agent = create_agent_with_knowledge()print("智能体初始化完成！")def chat_with_agent(message, history):
    """处理对话"""
    try:
        response = agent.chat(message)
        return response    except Exception as e:
        return f"抱歉，处理时出现错误：{str(e)}"# 创建Gradio界面with gr.Blocks(title="DeepSeek智能体知识库", theme=gr.themes.Soft()) as demo:
    gr.Markdown("""
    # 					

										

  本站不存储任何实质资源，该帖为网盘用户发布的网盘链接介绍帖,本文内所有链接指向的云盘网盘资源，其版权归版权方所有！其实际管理权为帖子发布者所有，本站无法操作相关资源。如您认为本站任何介绍帖侵犯了您的合法版权，请发送邮件 [email protected] 进行投诉，我们将在确认本文链接指向的资源存在侵权后，立即删除相关介绍帖子！