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