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踩过这些坑并对系统进行针对性优化后你的智能体应用将从一个脆弱的原型进化成一个健壮、可用、成本可控的生产级服务雏形。记住迭代和测试是关键每增加一个功能或修改一处逻辑都要用尽可能多的场景去验证它。
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踩过这些坑并对系统进行针对性优化后你的智能体应用将从一个脆弱的原型进化成一个健壮、可用、成本可控的生产级服务雏形。记住迭代和测试是关键每增加一个功能或修改一处逻辑都要用尽可能多的场景去验证它。
