1. 項目概述從零構建你的第一個智能體應用最近在跟幾個做AI應用的朋友聊天發現大家討論的焦點已經從“怎么調大模型API”轉向了“怎么讓大模型真正干點復雜的活兒”。比如讓AI自動分析一份幾十頁的PDF報告然后根據分析結果去數據庫里查數據最后生成一份帶圖表的周報。這種需要多步驟、有狀態、能自主決策的應用就是現在常說的“智能體”。聽起來很酷但新手一上手就容易懵Agent、RAG、LangGraph這些詞到底啥關系代碼從哪開始寫這正是我們接下來15天要一起解決的問題。這個系列不是理論課而是一個完整的、手把手的代碼實操項目。我們的目標很明確從零開始用Python和FastAPI搭建一個具備長期記憶和復雜工作流能力的原生智能體應用。你會親手實現一個能理解你問題、從自己的知識庫RAG里找答案、并能按步驟執行任務LangGraph的AI助手。無論你是剛學完Python基礎想找項目練手還是已經用過LangChain但想更深入底層原理這個系列都能給你帶來實實在在的代碼和思路。2. 核心概念拆解Agent、RAG與LangGraph為何是黃金三角在動手寫代碼之前我們必須先理清這三個核心概念各自扮演什么角色以及它們如何協同工作。很多人容易把它們混為一談其實它們分工非常明確。2.1 智能體從“問答機”到“執行者”的蛻變傳統的聊天機器人你問它答一次交互就結束了它不記得之前說過什么也不會主動去做事。智能體則是一個更高級的概念。你可以把它想象成一個虛擬的、擁有一定自主權的員工。它不僅有“大腦”大語言模型還有“手”和“眼睛”工具集更重要的是它有“工作流程”和“記憶”。一個典型的智能體工作循環是接收你的指令 - 思考決定用什么工具、怎么分解任務- 執行調用搜索、計算、寫代碼等工具- 觀察結果 - 再思考 - 直到任務完成或無法繼續。這個“思考-行動-觀察”的循環是智能體區別于簡單問答的核心。在代碼層面智能體通常由一個“大腦”LLM和一個“工具調用框架”組成它負責決策和調度。2.2 RAG為智能體裝上“長期記憶”與“專業手冊”大模型很聰明但它有兩個致命弱點知識可能過時以及會產生“幻覺”一本正經地胡說八道。比如你問它公司內部最新的銷售政策它肯定不知道。RAG就是為了解決這個問題而生的。它的全稱是“檢索增強生成”原理很像一個學霸考試先不急著答題而是快速翻閱允許帶進考場的參考資料檢索找到相關段落然后結合這些資料和自己的知識組織答案增強生成。在技術實現上RAG分為三步索引把你的文檔PDF、Word、網頁等切分成片段轉換成向量存入向量數據庫如Milvus、Chroma。檢索當用戶提問時將問題也轉換成向量在數據庫中找出最相似的幾個文本片段。生成把這些片段作為上下文連同問題一起送給大模型讓它生成基于這些事實的答案。這樣智能體就擁有了一個隨時可查、私有的、最新的知識庫回答專業問題的準確率會大幅提升。2.3 LangGraph為智能體設計“工作流程圖”智能體的任務往往不是一步到位的。比如“幫我分析上周銷售數據并寫郵件給經理”這至少包含取數據、分析、生成報告、起草郵件等多個步驟步驟間可能有條件分支如果銷售額下降則分析原因如果上升則總結經驗。用傳統的if-else寫這種流程代碼會很快變成一團亂麻。LangGraph就是一個專門用來描述和運行這種有狀態、可循環、帶分支的工作流的庫。它用“圖”的概念來建模節點代表一個步驟如調用LLM、執行工具邊代表步驟之間的流轉條件。它的核心價值在于讓復雜的工作流變得清晰、可維護、可可視化。你可以明確地看到任務從“開始”節點經過“決策”節點根據結果走不同的“分支”最終到達“結束”節點。LangGraph是LangChain生態系統的一部分但更專注于復雜控制流。三者關系總結RAG是智能體的“知識庫”和“記憶體”讓它的回答有據可依LangGraph是智能體的“流程引擎”和“調度中心”讓它能處理復雜任務而智能體自身則是整合這一切的“大腦”和“執行主體”。我們這個項目就是要將它們有機地組合在一起。3. 環境搭建與基礎工具鏈配置工欲善其事必先利其器。我們選擇Python作為主要語言因為它擁有最豐富的AI生態。下面是一份詳細的、避坑的環境配置指南。3.1 Python與包管理工具避免環境沖突的基石首先強烈建議使用Miniconda或Anaconda來創建獨立的虛擬環境。這能保證項目依賴不會污染你的系統Python也方便不同項目使用不同版本的包。如果你已經安裝了Python可以通過python --version檢查版本推薦使用Python 3.9或3.10穩定性最好。# 創建名為ai_agent的虛擬環境指定Python版本 conda create -n ai_agent python3.10 -y # 激活環境 conda activate ai_agent接下來是包管理。除了經典的pip我強烈推薦使用uv或pdm作為新的包管理工具。它們速度極快能生成精確的鎖文件徹底解決“在我機器上好好的”這種問題。這里以uv為例需先安裝pip install uv。# 在項目根目錄初始化這會生成pyproject.toml文件 uv init # 添加核心依賴uv會處理依賴解析和安裝速度比pip快很多 uv add openai langchain langchain-openai langgraph chromadb pypdf fastapi uvicorn注意網絡問題是環境配置的第一大敵。如果你在安裝某些包特別是涉及TensorFlow或某些底層C庫的時遇到超時或失敗請優先考慮更換pip源為國內鏡像如清華源、阿里云源。對于uv可以通過環境變量設置export UV_INDEX_URLhttps://pypi.tuna.tsinghua.edu.cn/simple。3.2 核心庫選型解析為什么是它們OpenAI / LangChain-OpenAI我們使用OpenAI的GPT系列模型作為智能體的“大腦”。langchain-openai是LangChain官方維護的集成包比直接用OpenAI SDK更方便與LangChain生態結合。LangChain LangGraph這是我們的核心框架。LangChain提供了構建鏈和智能體所需的大量組件提示模板、輸出解析器、記憶等而LangGraph則用于構建復雜工作流。注意我們雖然用LangChain但本系列會側重于講解其原理并嘗試部分“原生”實現以加深理解。ChromaDB一個輕量級、易用的開源向量數據庫非常適合本地開發和中小型項目。我們將用它來存儲文檔向量實現RAG的檢索功能。FastAPI UvicornFastAPI是一個現代、高性能的Python Web框架非常適合構建AI應用的API接口。Uvicorn是一個快速的ASGI服務器用于運行FastAPI應用。3.3 初始化第一個智能體與LLM的第一次對話環境準備好后我們來寫第一個腳本驗證一切是否正常并實現最簡單的問答。首先你需要準備一個OpenAI的API Key。請妥善保管不要直接硬編碼在代碼里。# 文件simple_agent.py import os from langchain_openai import ChatOpenAI # 方法1設置環境變量推薦 os.environ[OPENAI_API_KEY] 你的-api-key-here # 方法2在初始化時傳入適用于多密鑰管理 llm ChatOpenAI( modelgpt-3.5-turbo, # 或 gpt-4 temperature0, # 控制創造性0表示最確定性的輸出 api_key你的-api-key-here # 如果不設置環境變量可以在這里傳 ) # 進行第一次對話 response llm.invoke(你好請用一句話介紹你自己。) print(response.content)運行這個腳本如果看到模型的自我介紹恭喜你智能體的“大腦”已經接通了。這里的ChatOpenAI對象就是對大語言模型的封裝。temperature參數很重要對于需要確定性答案的任務如代碼生成、數據提取設為0或接近0的值對于需要創造性的任務如寫故事、頭腦風暴可以設為0.7~1.0。4. 構建你的第一個RAG知識庫系統有了會思考的大腦接下來我們給它裝備一個私人圖書館。我們將創建一個完整的RAG系統實現文檔上傳、向量化存儲和智能檢索回答。4.1 文檔加載與預處理從PDF到文本片段RAG的第一步是把非結構化的文檔變成結構化的、可檢索的文本塊。這里以PDF為例我們使用PyPDF2或pypdf。# 文件rag_ingest.py from langchain_community.document_loaders import PyPDFLoader from langchain_text_splitters import RecursiveCharacterTextSplitter # 1. 加載文檔 loader PyPDFLoader(./data/your_document.pdf) # 假設你的PDF放在data文件夾下 documents loader.load() print(f加載了 {len(documents)} 頁文檔。) # 2. 分割文本 text_splitter RecursiveCharacterTextSplitter( chunk_size500, # 每個文本塊的最大字符數 chunk_overlap50, # 塊與塊之間的重疊字符數防止上下文斷裂 length_functionlen, separators[\n\n, \n, 。, , , , , , ] # 分割符優先級 ) chunks text_splitter.split_documents(documents) print(f將文檔切分成了 {len(chunks)} 個文本塊。) # 查看第一個塊的內容和元數據 print(示例塊內容:, chunks[0].page_content[:200]) print(示例塊元數據:, chunks[0].metadata)實操心得chunk_size和chunk_overlap是需要反復調試的關鍵參數。尺寸太小會丟失上下文太大會引入無關噪聲并增加檢索成本。對于普通技術文檔500-1000是個不錯的起點。重疊部分能有效避免一個完整的句子或概念被攔腰切斷。4.2 向量化與存儲將文本轉換為可計算的距離文本塊需要轉換成向量一組數字才能進行相似度計算。我們使用OpenAI的文本嵌入模型并將向量存入ChromaDB。# 文件rag_ingest.py (續) from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma # 3. 創建嵌入模型 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) # 性價比高效果足夠 # 4. 創建向量數據庫并持久化 vectorstore Chroma.from_documents( documentschunks, embeddingembeddings, persist_directory./chroma_db # 指定持久化目錄 ) vectorstore.persist() # 顯式保存到磁盤 print(向量數據庫已創建并保存到 ./chroma_db 目錄。)嵌入模型將每個文本塊轉換為一個1536維對于text-embedding-3-small的向量。這個向量就像文本在“語義空間”中的坐標語義相近的文本其向量的“距離”通常用余弦相似度衡量也更近。4.3 檢索與問答鏈實現基于知識的回答知識庫建好了現在來實現問答功能。核心是“檢索器”和“問答鏈”。# 文件rag_query.py from langchain_chroma import Chroma from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate # 1. 加載已存在的向量數據庫 embeddings OpenAIEmbeddings() vectorstore Chroma( persist_directory./chroma_db, embedding_functionembeddings ) # 2. 創建檢索器。search_kwargs可以控制返回的相似文本塊數量 retriever vectorstore.as_retriever(search_kwargs{k: 3}) # 3. 自定義提示模板讓模型更好地利用上下文 prompt_template 請根據以下上下文信息來回答問題。如果你不知道答案就說你不知道不要編造答案。 上下文 {context} 問題{question} 請給出詳細的回答 PROMPT PromptTemplate( templateprompt_template, input_variables[context, question] ) # 4. 創建檢索問答鏈 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, # 最簡單的方式將所有檢索到的上下文塞入提示 retrieverretriever, chain_type_kwargs{prompt: PROMPT}, # 使用自定義提示 return_source_documentsTrue # 返回來源文檔便于溯源 ) # 5. 進行提問 question 文檔中提到的核心挑戰是什么 result qa_chain.invoke({query: question}) print(答案, result[result]) print(\n--- 來源文檔 ---) for i, doc in enumerate(result[source_documents]): print(f[來源{i1}] {doc.page_content[:150]}...)現在你的智能體已經能夠根據你提供的私有文檔來回答問題并且答案有據可查。chain_typestuff是最直接的方式但對于大量檢索結果可能會超出模型上下文長度。對于更長的文檔可以考慮map_reduce或refine等更復雜的鏈類型。5. 深入LangGraph設計智能體的工作流引擎RAG讓智能體有了知識LangGraph則賦予它執行復雜任務的能力。我們從一個簡單的“研究助手”智能體開始它需要判斷用戶問題是否需要聯網搜索。5.1 定義狀態與節點工作流的基石在LangGraph中一切圍繞“狀態”和“節點”進行。狀態是一個字典存儲工作流執行過程中的所有數據。節點是一個函數接收狀態執行操作并返回更新后的狀態。# 文件langgraph_agent.py from typing import TypedDict, Annotated, List from langgraph.graph import StateGraph, END import operator # 1. 定義狀態結構。這就像工作流的“共享白板”。 class AgentState(TypedDict): question: str # 用戶原始問題 needs_search: bool # 是否需要聯網搜索 search_results: str # 搜索到的結果 final_answer: str # 最終答案 # 2. 定義節點函數 def decide_search_node(state: AgentState) - AgentState: 決策節點判斷問題是否需要聯網搜索。 from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 構建一個分類提示 classification_prompt f 請判斷以下問題是否需要通過聯網搜索最新信息來回答。 如果問題是關于實時信息、新聞、股價、天氣、或2023年7月之后發生的特定事件請回答“是”。 如果問題基于通用知識、歷史事實、或文檔內容即可回答請回答“否”。 問題{state[question]} 只需回答“是”或“否”。 response llm.invoke(classification_prompt) needs_search response.content.strip() 是 # 更新狀態 return {needs_search: needs_search} def web_search_node(state: AgentState) - AgentState: 搜索節點模擬聯網搜索。 # 注意這里為了演示模擬搜索。實際中應集成SerpAPI、Tavily等真實搜索工具。 if state[needs_search]: print(f正在搜索: {state[question]}) # 模擬搜索返回結果 mock_results f關于{state[question]}的模擬搜索結果當前信息為XXX。 return {search_results: mock_results} else: # 如果不需要搜索直接傳遞空結果 return {search_results: 無需搜索。} def answer_node(state: AgentState) - AgentState: 回答節點綜合所有信息生成最終答案。 from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 根據是否有搜索結果構建不同的提示 if state[needs_search] and state[search_results]: prompt f 請基于以下搜索結果為用戶的問題提供一個全面、準確的答案。 用戶問題{state[question]} 搜索結果{state[search_results]} 請整合信息給出最終答案 else: # 這里可以集成之前構建的RAG系統例如 # answer qa_chain.invoke({query: state[question]}) # final answer[result] # 為了演示我們先使用通用模型回答 prompt f 請回答以下問題。如果你不知道請如實說明。 問題{state[question]} 答案 response llm.invoke(prompt) return {final_answer: response.content}5.2 構建與運行圖讓工作流動起來定義了節點后我們需要用邊把它們連接起來形成一個有向圖。# 文件langgraph_agent.py (續) # 3. 創建圖構建器 workflow StateGraph(AgentState) # 4. 添加節點 workflow.add_node(decide_search, decide_search_node) workflow.add_node(web_search, web_search_node) workflow.add_node(generate_answer, answer_node) # 5. 添加邊定義流程 workflow.set_entry_point(decide_search) # 設置入口節點 # 從決策節點出發根據狀態中的needs_search值決定下一步 workflow.add_conditional_edges( decide_search, # 這是一個路由函數根據當前狀態返回下一個節點的名稱 lambda state: web_search if state[needs_search] else generate_answer, { web_search: web_search, # 如果返回“web_search”則去web_search節點 generate_answer: generate_answer # 否則直接去回答節點 } ) # 設置無條件邊 workflow.add_edge(web_search, generate_answer) # 搜索完一定去回答 workflow.add_edge(generate_answer, END) # 回答完就結束 # 6. 編譯圖 app workflow.compile() # 7. 運行圖 initial_state AgentState(question今天北京的天氣怎么樣) result app.invoke(initial_state) print(最終答案, result[final_answer]) print(完整狀態, result)這個簡單的圖包含了條件判斷。你可以通過app.get_graph().draw_mermaid_png()輸出流程圖需要安裝pygraphviz直觀地看到decide_search - (web_search - generate_answer - END)或decide_search - generate_answer - END兩條路徑。5.3 集成工具調用讓智能體真正“動手”上面的搜索節點是模擬的。真正的智能體需要調用外部工具。我們來集成一個計算器和真實的搜索API以模擬為例。# 文件tool_agent.py from langchain.tools import tool from langchain_openai import ChatOpenAI from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain.prompts import ChatPromptTemplate # 1. 定義工具。使用tool裝飾器。 tool def calculate(expression: str) - str: 計算一個數學表達式。例如calculate(23*4)。 try: # 警告使用eval有安全風險僅用于演示。生產環境應用ast.literal_eval或專用庫。 result eval(expression) return f計算結果{result} except Exception as e: return f計算錯誤{e} tool def search_web(query: str) - str: 在網絡上搜索信息。 # 模擬搜索返回 return f模擬搜索{query}的結果相關信息是... # 2. 準備工具列表和LLM tools [calculate, search_web] llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 3. 創建智能體提示模板 prompt ChatPromptTemplate.from_messages([ (system, 你是一個有幫助的助手可以調用工具來回答問題。), (placeholder, {chat_history}), (human, {input}), (placeholder, {agent_scratchpad}), ]) # 4. 創建智能體和執行器 agent create_tool_calling_agent(llm, tools, prompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue) # verboseTrue 打印思考過程 # 5. 運行智能體 result agent_executor.invoke({input: 先計算(15的平方根是多少)然后搜索一下最新的AI新聞。}) print(result[output])當你運行這段代碼并設置verboseTrue時會在控制臺看到智能體的完整思考過程它先決定調用calculate工具得到結果后再決定調用search_web工具最后整合信息給出回答。這就是智能體“思考-行動-觀察”循環的直觀體現。6. 項目實戰搭建一個具備長期記憶的FastAPI智能體服務現在我們將前面所有模塊整合起來構建一個可以通過HTTP API訪問的、具備RAG知識庫和復雜工作流的智能體服務。6.1 使用FastAPI構建API端點我們將創建兩個主要端點一個用于上傳文檔到知識庫一個用于向智能體提問。# 文件main.py from fastapi import FastAPI, File, UploadFile, HTTPException from fastapi.responses import JSONResponse from pydantic import BaseModel import os import shutil from typing import List # 導入我們之前寫的RAG和Agent函數需要稍作調整封裝成函數 from rag_ingest import ingest_document_to_vectorstore from rag_query import get_qa_chain from langgraph_agent import get_agent_app app FastAPI(title智能體API服務) # 全局變量生產環境應使用數據庫或緩存 vector_store None agent_app None class QueryRequest(BaseModel): question: str use_agent: bool True # 是否使用LangGraph智能體工作流 app.on_event(startup) async def startup_event(): 服務啟動時加載已有的向量庫和智能體圖。 global vector_store, agent_app try: # 加載RAG問答鏈內部會加載向量庫 vector_store get_qa_chain() print(RAG向量庫加載成功。) except Exception as e: print(f加載RAG向量庫失敗將僅使用智能體功能: {e}) vector_store None # 初始化LangGraph智能體 agent_app get_agent_app(vector_store) # 假設我們修改了函數能接收RAG鏈 print(智能體圖編譯成功。) app.post(/upload/) async def upload_document(file: UploadFile File(...)): 上傳文檔并添加到知識庫。 if not file.filename.endswith(.pdf): raise HTTPException(status_code400, detail僅支持PDF文件。) # 保存上傳的文件 file_path f./uploads/{file.filename} os.makedirs(os.path.dirname(file_path), exist_okTrue) with open(file_path, wb) as buffer: shutil.copyfileobj(file.file, buffer) try: # 調用 ingest 函數處理文檔 # 注意這里需要更新全局的vector_store實際項目應考慮線程安全或重新加載 ingest_document_to_vectorstore(file_path) # 簡化處理提示用戶需要重啟服務或設計動態加載邏輯 return JSONResponse(content{message: f文檔{file.filename}已接收知識庫更新需重啟服務或調用特定接口。}) except Exception as e: raise HTTPException(status_code500, detailf文檔處理失敗: {str(e)}) app.post(/query/) async def query_agent(request: QueryRequest): 向智能體提問。 global vector_store, agent_app if agent_app is None: raise HTTPException(status_code500, detail智能體未初始化。) try: # 構建初始狀態 initial_state { question: request.question, use_rag: vector_store is not None, # ... 其他初始狀態 } # 運行智能體圖 result agent_app.invoke(initial_state) return JSONResponse(content{ answer: result.get(final_answer, 未生成答案), intermediate_steps: result # 可以過濾返回關鍵步驟 }) except Exception as e: raise HTTPException(status_code500, detailf智能體執行出錯: {str(e)}) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)6.2 集成RAG與LangGraph狀態共享與路由決策我們需要改造之前的langgraph_agent.py使其能根據情況決定是使用RAG知識庫還是聯網搜索或者兩者結合。# 文件advanced_agent.py (部分關鍵代碼) def router_node(state: AgentState): 路由節點決定使用哪種信息源。 question state[question] # 這里可以實現更復雜的路由邏輯例如 # 1. 用一個小模型判斷問題類型事實型、計算型、創意型... # 2. 檢查問題是否涉及私有知識通過關鍵詞或嵌入相似度初步判斷 # 簡化版如果問題包含“文檔”、“報告”、“根據資料”等詞優先用RAG rag_keywords [文檔, 資料, 報告, 文中] if any(keyword in question for keyword in rag_keywords) and state.get(rag_chain): return {next_step: use_rag} elif state[needs_search]: # 之前的判斷邏輯 return {next_step: use_search} else: return {next_step: use_llm_only} def rag_answering_node(state: AgentState): 調用RAG鏈回答。 rag_chain state[rag_chain] result rag_chain.invoke({query: state[question]}) return {rag_answer: result[result], source_docs: result[source_documents]} # 在圖中添加條件邊根據next_step路由到不同的回答節點。6.3 部署與測試讓服務跑起來安裝依賴確保在虛擬環境中安裝了所有包fastapi,uvicorn,python-multipart用于文件上傳。運行服務在項目根目錄執行python main.py或uvicorn main:app --reload --host 0.0.0.0 --port 8000。測試API打開瀏覽器訪問http://127.0.0.1:8000/docs你會看到自動生成的交互式API文檔Swagger UI。在/upload/端點嘗試上傳一個PDF文件。在/query/端點嘗試提問。{ question: 根據你已學習的文檔總結核心要點。, use_agent: true }7. 避坑指南與性能優化實戰在實際開發和運行中你會遇到各種各樣的問題。這里記錄了一些常見的“坑”和優化思路。7.1 常見錯誤與排查清單錯誤現象可能原因排查步驟ModuleNotFoundError依賴未安裝或虛擬環境未激活1. 確認已激活正確的conda/venv環境。2. 運行pip list或uv pip list檢查關鍵包是否存在。3. 在PyCharm/VSCode中檢查項目解釋器設置。OpenAI API調用超時或報錯網絡問題、API密鑰錯誤、額度不足1. 檢查網絡連接特別是代理設置。2. 驗證API Key是否正確且有效。3. 登錄OpenAI平臺檢查用量和余額。ChromaDB報persist相關錯誤目錄權限問題、舊版本不兼容1. 確保程序對./chroma_db目錄有讀寫權限。2. 嘗試刪除舊的chroma_db文件夾重新生成。3. 升級ChromaDB到最新版本。RAG回答質量差答非所問文本分割不合理、檢索數量k值不當、提示詞不佳1. 檢查文本分割后的塊看是否語義完整。2. 調整chunk_size和chunk_overlap。3. 增加或減少檢索數量k通常3-5。4. 優化提示模板明確指令“根據上下文回答”。LangGraph圖編譯或運行出錯狀態結構定義與節點返回值不匹配、邊未正確連接1. 檢查State的TypedDict定義是否包含所有節點可能更新的鍵。2. 確保每個節點返回的字典是狀態鍵的子集。3. 使用app.get_graph().draw_mermaid_png()可視化檢查圖結構。智能體頻繁調用錯誤工具工具描述不清晰、LLM溫度過高1. 為每個tool編寫清晰、具體的描述說明輸入輸出。2. 將LLM的temperature調低如0增加確定性。3. 在系統提示詞中強調“必須使用提供的工具”。7.2 性能優化與成本控制1. 嵌入模型的選擇本地模型如BAAI/bge-small-zh無需API調用零成本適合中文或對延遲敏感的內部應用。可使用langchain_huggingface集成。小型API模型OpenAI的text-embedding-3-small在成本、速度和效果間取得了很好平衡是云端應用的默認選擇。成本計算假設文檔有1000個塊每個塊500字符。使用text-embedding-3-small每1K tokens $0.00002嵌入成本約為1000 * (500/4) / 1000 * $0.00002 ≈ $0.0025非常低廉。查詢成本類似。2. 檢索優化分層索引先使用簡單的關鍵詞匹配如BM25快速篩選出一批文檔再對這批文檔用向量檢索做精排兼顧速度和精度。元數據過濾在檢索時加入過濾條件如文檔類型、日期、作者等可以大幅提升檢索準確率。ChromaDB支持此功能。重排序檢索出Top K個結果如K20后使用一個更小、更快的重排序模型對它們進行精排再取Top N如N3送入LLM效果提升顯著。3. 智能體流程優化減少不必要的LLM調用在路由節點可以用規則或小模型先做粗篩避免每個問題都調用大模型做決策。設置超時與重試對于工具調用如網絡請求務必設置超時并實現簡單的重試邏輯提高系統健壯性。流式輸出對于長文本生成使用FastAPI的StreamingResponse和LangChain的流式回調實現逐詞輸出提升用戶體驗。4. 異步處理 對于API服務使用異步框架FastAPI本身支持異步和異步的LangChain組件如AsyncChromaLangChain可以顯著提高并發吞吐量避免在I/O操作如LLM API調用、數據庫查詢時阻塞整個服務。# 示例在FastAPI中異步調用智能體 app.post(/async_query/) async def async_query(request: QueryRequest): # 注意需要確保你使用的LangChain組件支持異步例如使用 ainvoke result await agent_app.ainvoke({input: request.question}) return result踩過這些坑并對系統進行針對性優化后你的智能體應用將從一個脆弱的原型進化成一個健壯、可用、成本可控的生產級服務雛形。記住迭代和測試是關鍵每增加一個功能或修改一處邏輯都要用盡可能多的場景去驗證它。 從零構建智能體應用：Agent、RAG與LangGraph實戰指南
1. 項目概述從零構建你的第一個智能體應用最近在跟幾個做AI應用的朋友聊天發現大家討論的焦點已經從“怎么調大模型API”轉向了“怎么讓大模型真正干點復雜的活兒”。比如讓AI自動分析一份幾十頁的PDF報告然后根據分析結果去數據庫里查數據最后生成一份帶圖表的周報。這種需要多步驟、有狀態、能自主決策的應用就是現在常說的“智能體”。聽起來很酷但新手一上手就容易懵Agent、RAG、LangGraph這些詞到底啥關系代碼從哪開始寫這正是我們接下來15天要一起解決的問題。這個系列不是理論課而是一個完整的、手把手的代碼實操項目。我們的目標很明確從零開始用Python和FastAPI搭建一個具備長期記憶和復雜工作流能力的原生智能體應用。你會親手實現一個能理解你問題、從自己的知識庫RAG里找答案、并能按步驟執行任務LangGraph的AI助手。無論你是剛學完Python基礎想找項目練手還是已經用過LangChain但想更深入底層原理這個系列都能給你帶來實實在在的代碼和思路。2. 核心概念拆解Agent、RAG與LangGraph為何是黃金三角在動手寫代碼之前我們必須先理清這三個核心概念各自扮演什么角色以及它們如何協同工作。很多人容易把它們混為一談其實它們分工非常明確。2.1 智能體從“問答機”到“執行者”的蛻變傳統的聊天機器人你問它答一次交互就結束了它不記得之前說過什么也不會主動去做事。智能體則是一個更高級的概念。你可以把它想象成一個虛擬的、擁有一定自主權的員工。它不僅有“大腦”大語言模型還有“手”和“眼睛”工具集更重要的是它有“工作流程”和“記憶”。一個典型的智能體工作循環是接收你的指令 - 思考決定用什么工具、怎么分解任務- 執行調用搜索、計算、寫代碼等工具- 觀察結果 - 再思考 - 直到任務完成或無法繼續。這個“思考-行動-觀察”的循環是智能體區別于簡單問答的核心。在代碼層面智能體通常由一個“大腦”LLM和一個“工具調用框架”組成它負責決策和調度。2.2 RAG為智能體裝上“長期記憶”與“專業手冊”大模型很聰明但它有兩個致命弱點知識可能過時以及會產生“幻覺”一本正經地胡說八道。比如你問它公司內部最新的銷售政策它肯定不知道。RAG就是為了解決這個問題而生的。它的全稱是“檢索增強生成”原理很像一個學霸考試先不急著答題而是快速翻閱允許帶進考場的參考資料檢索找到相關段落然后結合這些資料和自己的知識組織答案增強生成。在技術實現上RAG分為三步索引把你的文檔PDF、Word、網頁等切分成片段轉換成向量存入向量數據庫如Milvus、Chroma。檢索當用戶提問時將問題也轉換成向量在數據庫中找出最相似的幾個文本片段。生成把這些片段作為上下文連同問題一起送給大模型讓它生成基于這些事實的答案。這樣智能體就擁有了一個隨時可查、私有的、最新的知識庫回答專業問題的準確率會大幅提升。2.3 LangGraph為智能體設計“工作流程圖”智能體的任務往往不是一步到位的。比如“幫我分析上周銷售數據并寫郵件給經理”這至少包含取數據、分析、生成報告、起草郵件等多個步驟步驟間可能有條件分支如果銷售額下降則分析原因如果上升則總結經驗。用傳統的if-else寫這種流程代碼會很快變成一團亂麻。LangGraph就是一個專門用來描述和運行這種有狀態、可循環、帶分支的工作流的庫。它用“圖”的概念來建模節點代表一個步驟如調用LLM、執行工具邊代表步驟之間的流轉條件。它的核心價值在于讓復雜的工作流變得清晰、可維護、可可視化。你可以明確地看到任務從“開始”節點經過“決策”節點根據結果走不同的“分支”最終到達“結束”節點。LangGraph是LangChain生態系統的一部分但更專注于復雜控制流。三者關系總結RAG是智能體的“知識庫”和“記憶體”讓它的回答有據可依LangGraph是智能體的“流程引擎”和“調度中心”讓它能處理復雜任務而智能體自身則是整合這一切的“大腦”和“執行主體”。我們這個項目就是要將它們有機地組合在一起。3. 環境搭建與基礎工具鏈配置工欲善其事必先利其器。我們選擇Python作為主要語言因為它擁有最豐富的AI生態。下面是一份詳細的、避坑的環境配置指南。3.1 Python與包管理工具避免環境沖突的基石首先強烈建議使用Miniconda或Anaconda來創建獨立的虛擬環境。這能保證項目依賴不會污染你的系統Python也方便不同項目使用不同版本的包。如果你已經安裝了Python可以通過python --version檢查版本推薦使用Python 3.9或3.10穩定性最好。# 創建名為ai_agent的虛擬環境指定Python版本 conda create -n ai_agent python3.10 -y # 激活環境 conda activate ai_agent接下來是包管理。除了經典的pip我強烈推薦使用uv或pdm作為新的包管理工具。它們速度極快能生成精確的鎖文件徹底解決“在我機器上好好的”這種問題。這里以uv為例需先安裝pip install uv。# 在項目根目錄初始化這會生成pyproject.toml文件 uv init # 添加核心依賴uv會處理依賴解析和安裝速度比pip快很多 uv add openai langchain langchain-openai langgraph chromadb pypdf fastapi uvicorn注意網絡問題是環境配置的第一大敵。如果你在安裝某些包特別是涉及TensorFlow或某些底層C庫的時遇到超時或失敗請優先考慮更換pip源為國內鏡像如清華源、阿里云源。對于uv可以通過環境變量設置export UV_INDEX_URLhttps://pypi.tuna.tsinghua.edu.cn/simple。3.2 核心庫選型解析為什么是它們OpenAI / LangChain-OpenAI我們使用OpenAI的GPT系列模型作為智能體的“大腦”。langchain-openai是LangChain官方維護的集成包比直接用OpenAI SDK更方便與LangChain生態結合。LangChain LangGraph這是我們的核心框架。LangChain提供了構建鏈和智能體所需的大量組件提示模板、輸出解析器、記憶等而LangGraph則用于構建復雜工作流。注意我們雖然用LangChain但本系列會側重于講解其原理并嘗試部分“原生”實現以加深理解。ChromaDB一個輕量級、易用的開源向量數據庫非常適合本地開發和中小型項目。我們將用它來存儲文檔向量實現RAG的檢索功能。FastAPI UvicornFastAPI是一個現代、高性能的Python Web框架非常適合構建AI應用的API接口。Uvicorn是一個快速的ASGI服務器用于運行FastAPI應用。3.3 初始化第一個智能體與LLM的第一次對話環境準備好后我們來寫第一個腳本驗證一切是否正常并實現最簡單的問答。首先你需要準備一個OpenAI的API Key。請妥善保管不要直接硬編碼在代碼里。# 文件simple_agent.py import os from langchain_openai import ChatOpenAI # 方法1設置環境變量推薦 os.environ[OPENAI_API_KEY] 你的-api-key-here # 方法2在初始化時傳入適用于多密鑰管理 llm ChatOpenAI( modelgpt-3.5-turbo, # 或 gpt-4 temperature0, # 控制創造性0表示最確定性的輸出 api_key你的-api-key-here # 如果不設置環境變量可以在這里傳 ) # 進行第一次對話 response llm.invoke(你好請用一句話介紹你自己。) print(response.content)運行這個腳本如果看到模型的自我介紹恭喜你智能體的“大腦”已經接通了。這里的ChatOpenAI對象就是對大語言模型的封裝。temperature參數很重要對于需要確定性答案的任務如代碼生成、數據提取設為0或接近0的值對于需要創造性的任務如寫故事、頭腦風暴可以設為0.7~1.0。4. 構建你的第一個RAG知識庫系統有了會思考的大腦接下來我們給它裝備一個私人圖書館。我們將創建一個完整的RAG系統實現文檔上傳、向量化存儲和智能檢索回答。4.1 文檔加載與預處理從PDF到文本片段RAG的第一步是把非結構化的文檔變成結構化的、可檢索的文本塊。這里以PDF為例我們使用PyPDF2或pypdf。# 文件rag_ingest.py from langchain_community.document_loaders import PyPDFLoader from langchain_text_splitters import RecursiveCharacterTextSplitter # 1. 加載文檔 loader PyPDFLoader(./data/your_document.pdf) # 假設你的PDF放在data文件夾下 documents loader.load() print(f加載了 {len(documents)} 頁文檔。) # 2. 分割文本 text_splitter RecursiveCharacterTextSplitter( chunk_size500, # 每個文本塊的最大字符數 chunk_overlap50, # 塊與塊之間的重疊字符數防止上下文斷裂 length_functionlen, separators[\n\n, \n, 。, , , , , , ] # 分割符優先級 ) chunks text_splitter.split_documents(documents) print(f將文檔切分成了 {len(chunks)} 個文本塊。) # 查看第一個塊的內容和元數據 print(示例塊內容:, chunks[0].page_content[:200]) print(示例塊元數據:, chunks[0].metadata)實操心得chunk_size和chunk_overlap是需要反復調試的關鍵參數。尺寸太小會丟失上下文太大會引入無關噪聲并增加檢索成本。對于普通技術文檔500-1000是個不錯的起點。重疊部分能有效避免一個完整的句子或概念被攔腰切斷。4.2 向量化與存儲將文本轉換為可計算的距離文本塊需要轉換成向量一組數字才能進行相似度計算。我們使用OpenAI的文本嵌入模型并將向量存入ChromaDB。# 文件rag_ingest.py (續) from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma # 3. 創建嵌入模型 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) # 性價比高效果足夠 # 4. 創建向量數據庫并持久化 vectorstore Chroma.from_documents( documentschunks, embeddingembeddings, persist_directory./chroma_db # 指定持久化目錄 ) vectorstore.persist() # 顯式保存到磁盤 print(向量數據庫已創建并保存到 ./chroma_db 目錄。)嵌入模型將每個文本塊轉換為一個1536維對于text-embedding-3-small的向量。這個向量就像文本在“語義空間”中的坐標語義相近的文本其向量的“距離”通常用余弦相似度衡量也更近。4.3 檢索與問答鏈實現基于知識的回答知識庫建好了現在來實現問答功能。核心是“檢索器”和“問答鏈”。# 文件rag_query.py from langchain_chroma import Chroma from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate # 1. 加載已存在的向量數據庫 embeddings OpenAIEmbeddings() vectorstore Chroma( persist_directory./chroma_db, embedding_functionembeddings ) # 2. 創建檢索器。search_kwargs可以控制返回的相似文本塊數量 retriever vectorstore.as_retriever(search_kwargs{k: 3}) # 3. 自定義提示模板讓模型更好地利用上下文 prompt_template 請根據以下上下文信息來回答問題。如果你不知道答案就說你不知道不要編造答案。 上下文 {context} 問題{question} 請給出詳細的回答 PROMPT PromptTemplate( templateprompt_template, input_variables[context, question] ) # 4. 創建檢索問答鏈 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, # 最簡單的方式將所有檢索到的上下文塞入提示 retrieverretriever, chain_type_kwargs{prompt: PROMPT}, # 使用自定義提示 return_source_documentsTrue # 返回來源文檔便于溯源 ) # 5. 進行提問 question 文檔中提到的核心挑戰是什么 result qa_chain.invoke({query: question}) print(答案, result[result]) print(\n--- 來源文檔 ---) for i, doc in enumerate(result[source_documents]): print(f[來源{i1}] {doc.page_content[:150]}...)現在你的智能體已經能夠根據你提供的私有文檔來回答問題并且答案有據可查。chain_typestuff是最直接的方式但對于大量檢索結果可能會超出模型上下文長度。對于更長的文檔可以考慮map_reduce或refine等更復雜的鏈類型。5. 深入LangGraph設計智能體的工作流引擎RAG讓智能體有了知識LangGraph則賦予它執行復雜任務的能力。我們從一個簡單的“研究助手”智能體開始它需要判斷用戶問題是否需要聯網搜索。5.1 定義狀態與節點工作流的基石在LangGraph中一切圍繞“狀態”和“節點”進行。狀態是一個字典存儲工作流執行過程中的所有數據。節點是一個函數接收狀態執行操作并返回更新后的狀態。# 文件langgraph_agent.py from typing import TypedDict, Annotated, List from langgraph.graph import StateGraph, END import operator # 1. 定義狀態結構。這就像工作流的“共享白板”。 class AgentState(TypedDict): question: str # 用戶原始問題 needs_search: bool # 是否需要聯網搜索 search_results: str # 搜索到的結果 final_answer: str # 最終答案 # 2. 定義節點函數 def decide_search_node(state: AgentState) - AgentState: 決策節點判斷問題是否需要聯網搜索。 from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 構建一個分類提示 classification_prompt f 請判斷以下問題是否需要通過聯網搜索最新信息來回答。 如果問題是關于實時信息、新聞、股價、天氣、或2023年7月之后發生的特定事件請回答“是”。 如果問題基于通用知識、歷史事實、或文檔內容即可回答請回答“否”。 問題{state[question]} 只需回答“是”或“否”。 response llm.invoke(classification_prompt) needs_search response.content.strip() 是 # 更新狀態 return {needs_search: needs_search} def web_search_node(state: AgentState) - AgentState: 搜索節點模擬聯網搜索。 # 注意這里為了演示模擬搜索。實際中應集成SerpAPI、Tavily等真實搜索工具。 if state[needs_search]: print(f正在搜索: {state[question]}) # 模擬搜索返回結果 mock_results f關于{state[question]}的模擬搜索結果當前信息為XXX。 return {search_results: mock_results} else: # 如果不需要搜索直接傳遞空結果 return {search_results: 無需搜索。} def answer_node(state: AgentState) - AgentState: 回答節點綜合所有信息生成最終答案。 from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 根據是否有搜索結果構建不同的提示 if state[needs_search] and state[search_results]: prompt f 請基于以下搜索結果為用戶的問題提供一個全面、準確的答案。 用戶問題{state[question]} 搜索結果{state[search_results]} 請整合信息給出最終答案 else: # 這里可以集成之前構建的RAG系統例如 # answer qa_chain.invoke({query: state[question]}) # final answer[result] # 為了演示我們先使用通用模型回答 prompt f 請回答以下問題。如果你不知道請如實說明。 問題{state[question]} 答案 response llm.invoke(prompt) return {final_answer: response.content}5.2 構建與運行圖讓工作流動起來定義了節點后我們需要用邊把它們連接起來形成一個有向圖。# 文件langgraph_agent.py (續) # 3. 創建圖構建器 workflow StateGraph(AgentState) # 4. 添加節點 workflow.add_node(decide_search, decide_search_node) workflow.add_node(web_search, web_search_node) workflow.add_node(generate_answer, answer_node) # 5. 添加邊定義流程 workflow.set_entry_point(decide_search) # 設置入口節點 # 從決策節點出發根據狀態中的needs_search值決定下一步 workflow.add_conditional_edges( decide_search, # 這是一個路由函數根據當前狀態返回下一個節點的名稱 lambda state: web_search if state[needs_search] else generate_answer, { web_search: web_search, # 如果返回“web_search”則去web_search節點 generate_answer: generate_answer # 否則直接去回答節點 } ) # 設置無條件邊 workflow.add_edge(web_search, generate_answer) # 搜索完一定去回答 workflow.add_edge(generate_answer, END) # 回答完就結束 # 6. 編譯圖 app workflow.compile() # 7. 運行圖 initial_state AgentState(question今天北京的天氣怎么樣) result app.invoke(initial_state) print(最終答案, result[final_answer]) print(完整狀態, result)這個簡單的圖包含了條件判斷。你可以通過app.get_graph().draw_mermaid_png()輸出流程圖需要安裝pygraphviz直觀地看到decide_search - (web_search - generate_answer - END)或decide_search - generate_answer - END兩條路徑。5.3 集成工具調用讓智能體真正“動手”上面的搜索節點是模擬的。真正的智能體需要調用外部工具。我們來集成一個計算器和真實的搜索API以模擬為例。# 文件tool_agent.py from langchain.tools import tool from langchain_openai import ChatOpenAI from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain.prompts import ChatPromptTemplate # 1. 定義工具。使用tool裝飾器。 tool def calculate(expression: str) - str: 計算一個數學表達式。例如calculate(23*4)。 try: # 警告使用eval有安全風險僅用于演示。生產環境應用ast.literal_eval或專用庫。 result eval(expression) return f計算結果{result} except Exception as e: return f計算錯誤{e} tool def search_web(query: str) - str: 在網絡上搜索信息。 # 模擬搜索返回 return f模擬搜索{query}的結果相關信息是... # 2. 準備工具列表和LLM tools [calculate, search_web] llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 3. 創建智能體提示模板 prompt ChatPromptTemplate.from_messages([ (system, 你是一個有幫助的助手可以調用工具來回答問題。), (placeholder, {chat_history}), (human, {input}), (placeholder, {agent_scratchpad}), ]) # 4. 創建智能體和執行器 agent create_tool_calling_agent(llm, tools, prompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue) # verboseTrue 打印思考過程 # 5. 運行智能體 result agent_executor.invoke({input: 先計算(15的平方根是多少)然后搜索一下最新的AI新聞。}) print(result[output])當你運行這段代碼并設置verboseTrue時會在控制臺看到智能體的完整思考過程它先決定調用calculate工具得到結果后再決定調用search_web工具最后整合信息給出回答。這就是智能體“思考-行動-觀察”循環的直觀體現。6. 項目實戰搭建一個具備長期記憶的FastAPI智能體服務現在我們將前面所有模塊整合起來構建一個可以通過HTTP API訪問的、具備RAG知識庫和復雜工作流的智能體服務。6.1 使用FastAPI構建API端點我們將創建兩個主要端點一個用于上傳文檔到知識庫一個用于向智能體提問。# 文件main.py from fastapi import FastAPI, File, UploadFile, HTTPException from fastapi.responses import JSONResponse from pydantic import BaseModel import os import shutil from typing import List # 導入我們之前寫的RAG和Agent函數需要稍作調整封裝成函數 from rag_ingest import ingest_document_to_vectorstore from rag_query import get_qa_chain from langgraph_agent import get_agent_app app FastAPI(title智能體API服務) # 全局變量生產環境應使用數據庫或緩存 vector_store None agent_app None class QueryRequest(BaseModel): question: str use_agent: bool True # 是否使用LangGraph智能體工作流 app.on_event(startup) async def startup_event(): 服務啟動時加載已有的向量庫和智能體圖。 global vector_store, agent_app try: # 加載RAG問答鏈內部會加載向量庫 vector_store get_qa_chain() print(RAG向量庫加載成功。) except Exception as e: print(f加載RAG向量庫失敗將僅使用智能體功能: {e}) vector_store None # 初始化LangGraph智能體 agent_app get_agent_app(vector_store) # 假設我們修改了函數能接收RAG鏈 print(智能體圖編譯成功。) app.post(/upload/) async def upload_document(file: UploadFile File(...)): 上傳文檔并添加到知識庫。 if not file.filename.endswith(.pdf): raise HTTPException(status_code400, detail僅支持PDF文件。) # 保存上傳的文件 file_path f./uploads/{file.filename} os.makedirs(os.path.dirname(file_path), exist_okTrue) with open(file_path, wb) as buffer: shutil.copyfileobj(file.file, buffer) try: # 調用 ingest 函數處理文檔 # 注意這里需要更新全局的vector_store實際項目應考慮線程安全或重新加載 ingest_document_to_vectorstore(file_path) # 簡化處理提示用戶需要重啟服務或設計動態加載邏輯 return JSONResponse(content{message: f文檔{file.filename}已接收知識庫更新需重啟服務或調用特定接口。}) except Exception as e: raise HTTPException(status_code500, detailf文檔處理失敗: {str(e)}) app.post(/query/) async def query_agent(request: QueryRequest): 向智能體提問。 global vector_store, agent_app if agent_app is None: raise HTTPException(status_code500, detail智能體未初始化。) try: # 構建初始狀態 initial_state { question: request.question, use_rag: vector_store is not None, # ... 其他初始狀態 } # 運行智能體圖 result agent_app.invoke(initial_state) return JSONResponse(content{ answer: result.get(final_answer, 未生成答案), intermediate_steps: result # 可以過濾返回關鍵步驟 }) except Exception as e: raise HTTPException(status_code500, detailf智能體執行出錯: {str(e)}) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)6.2 集成RAG與LangGraph狀態共享與路由決策我們需要改造之前的langgraph_agent.py使其能根據情況決定是使用RAG知識庫還是聯網搜索或者兩者結合。# 文件advanced_agent.py (部分關鍵代碼) def router_node(state: AgentState): 路由節點決定使用哪種信息源。 question state[question] # 這里可以實現更復雜的路由邏輯例如 # 1. 用一個小模型判斷問題類型事實型、計算型、創意型... # 2. 檢查問題是否涉及私有知識通過關鍵詞或嵌入相似度初步判斷 # 簡化版如果問題包含“文檔”、“報告”、“根據資料”等詞優先用RAG rag_keywords [文檔, 資料, 報告, 文中] if any(keyword in question for keyword in rag_keywords) and state.get(rag_chain): return {next_step: use_rag} elif state[needs_search]: # 之前的判斷邏輯 return {next_step: use_search} else: return {next_step: use_llm_only} def rag_answering_node(state: AgentState): 調用RAG鏈回答。 rag_chain state[rag_chain] result rag_chain.invoke({query: state[question]}) return {rag_answer: result[result], source_docs: result[source_documents]} # 在圖中添加條件邊根據next_step路由到不同的回答節點。6.3 部署與測試讓服務跑起來安裝依賴確保在虛擬環境中安裝了所有包fastapi,uvicorn,python-multipart用于文件上傳。運行服務在項目根目錄執行python main.py或uvicorn main:app --reload --host 0.0.0.0 --port 8000。測試API打開瀏覽器訪問http://127.0.0.1:8000/docs你會看到自動生成的交互式API文檔Swagger UI。在/upload/端點嘗試上傳一個PDF文件。在/query/端點嘗試提問。{ question: 根據你已學習的文檔總結核心要點。, use_agent: true }7. 避坑指南與性能優化實戰在實際開發和運行中你會遇到各種各樣的問題。這里記錄了一些常見的“坑”和優化思路。7.1 常見錯誤與排查清單錯誤現象可能原因排查步驟ModuleNotFoundError依賴未安裝或虛擬環境未激活1. 確認已激活正確的conda/venv環境。2. 運行pip list或uv pip list檢查關鍵包是否存在。3. 在PyCharm/VSCode中檢查項目解釋器設置。OpenAI API調用超時或報錯網絡問題、API密鑰錯誤、額度不足1. 檢查網絡連接特別是代理設置。2. 驗證API Key是否正確且有效。3. 登錄OpenAI平臺檢查用量和余額。ChromaDB報persist相關錯誤目錄權限問題、舊版本不兼容1. 確保程序對./chroma_db目錄有讀寫權限。2. 嘗試刪除舊的chroma_db文件夾重新生成。3. 升級ChromaDB到最新版本。RAG回答質量差答非所問文本分割不合理、檢索數量k值不當、提示詞不佳1. 檢查文本分割后的塊看是否語義完整。2. 調整chunk_size和chunk_overlap。3. 增加或減少檢索數量k通常3-5。4. 優化提示模板明確指令“根據上下文回答”。LangGraph圖編譯或運行出錯狀態結構定義與節點返回值不匹配、邊未正確連接1. 檢查State的TypedDict定義是否包含所有節點可能更新的鍵。2. 確保每個節點返回的字典是狀態鍵的子集。3. 使用app.get_graph().draw_mermaid_png()可視化檢查圖結構。智能體頻繁調用錯誤工具工具描述不清晰、LLM溫度過高1. 為每個tool編寫清晰、具體的描述說明輸入輸出。2. 將LLM的temperature調低如0增加確定性。3. 在系統提示詞中強調“必須使用提供的工具”。7.2 性能優化與成本控制1. 嵌入模型的選擇本地模型如BAAI/bge-small-zh無需API調用零成本適合中文或對延遲敏感的內部應用。可使用langchain_huggingface集成。小型API模型OpenAI的text-embedding-3-small在成本、速度和效果間取得了很好平衡是云端應用的默認選擇。成本計算假設文檔有1000個塊每個塊500字符。使用text-embedding-3-small每1K tokens $0.00002嵌入成本約為1000 * (500/4) / 1000 * $0.00002 ≈ $0.0025非常低廉。查詢成本類似。2. 檢索優化分層索引先使用簡單的關鍵詞匹配如BM25快速篩選出一批文檔再對這批文檔用向量檢索做精排兼顧速度和精度。元數據過濾在檢索時加入過濾條件如文檔類型、日期、作者等可以大幅提升檢索準確率。ChromaDB支持此功能。重排序檢索出Top K個結果如K20后使用一個更小、更快的重排序模型對它們進行精排再取Top N如N3送入LLM效果提升顯著。3. 智能體流程優化減少不必要的LLM調用在路由節點可以用規則或小模型先做粗篩避免每個問題都調用大模型做決策。設置超時與重試對于工具調用如網絡請求務必設置超時并實現簡單的重試邏輯提高系統健壯性。流式輸出對于長文本生成使用FastAPI的StreamingResponse和LangChain的流式回調實現逐詞輸出提升用戶體驗。4. 異步處理 對于API服務使用異步框架FastAPI本身支持異步和異步的LangChain組件如AsyncChromaLangChain可以顯著提高并發吞吐量避免在I/O操作如LLM API調用、數據庫查詢時阻塞整個服務。# 示例在FastAPI中異步調用智能體 app.post(/async_query/) async def async_query(request: QueryRequest): # 注意需要確保你使用的LangChain組件支持異步例如使用 ainvoke result await agent_app.ainvoke({input: request.question}) return result踩過這些坑并對系統進行針對性優化后你的智能體應用將從一個脆弱的原型進化成一個健壯、可用、成本可控的生產級服務雛形。記住迭代和測試是關鍵每增加一個功能或修改一處邏輯都要用盡可能多的場景去驗證它。