1. 从零到一为什么我们需要Agent、RAG与LangGraph如果你最近在AI应用开发圈子里待过大概率已经被“Agent”、“RAG”这些词刷屏了。但说实话很多教程要么上来就甩给你一堆代码要么就是大谈特谈概念看完之后感觉懂了一动手就懵。这就像有人告诉你“开车很简单踩油门就走”但没告诉你离合器、档位和交通规则。我花了半个月时间从零开始把原生Agent、RAG和LangGraph这三个核心概念揉在一起用代码实操了一遍。这个过程里踩的坑、绕的弯以及最终跑通一个能“思考”、能“查资料”、能“按计划行事”的智能体的快感我想完整地分享给你。简单来说我们想解决的问题是让AI模型比如GPT从一个“知识截止到2023年初、只会被动回答问题”的聊天机器人进化成一个能主动使用工具、访问最新或私有知识、并按照复杂逻辑执行多步骤任务的智能助手。这就是Agent智能体的愿景。而实现它离不开两样东西RAG检索增强生成给AI装上“外部记忆库”LangGraph则提供了编排这些复杂任务的“流程图”和“状态机”。接下来这15天的内容我会带你一步步搭建环境理解核心原理并用完整的代码实现一个具备长期记忆、能调用工具、可规划任务链的AI Agent。无论你是刚学完Python基础想找项目练手还是已经有一定经验想深入AI应用层开发这个系列都能给你一条清晰的路径。2. 环境搭建与核心库全景图不只是pip install工欲善其事必先利其器。在开始写第一行Agent代码之前一个稳定、可复现的开发环境至关重要。很多人在这里就会遇到第一个拦路虎。2.1 Python与包管理避开版本地狱首先确保你安装了Python 3.10或以上版本。这是大多数现代AI库的基线要求。我强烈建议使用conda或venv创建独立的虚拟环境这是避免包冲突的生命线。# 使用conda推荐尤其对Windows用户友好 conda create -n ai-agent python3.10 conda activate ai-agent # 或者使用venv python -m venv ai-agent # Windows激活 ai-agent\Scripts\activate # Mac/Linux激活 source ai-agent/bin/activate接下来是安装核心库。别急着一次性pip install所有东西我们分层进行。# 1. 基础框架层LangChain和LangGraph # LangChain是构建AI应用链式的工具箱LangGraph是它的扩展用于构建有状态的、循环的图即Agent的核心 pip install langchain langgraph # 2. 大模型接口层我们需要一个“大脑”。这里以OpenAI为例你也可以用Anthropic、Ollama本地模型等。 pip install openai # 3. 向量数据库与嵌入层这是RAG的“记忆库”部分。我们选用轻量且流行的ChromaDB。 pip install chromadb # 4. Web框架与工具层为了让Agent能对外提供服务或调用外部API我们引入FastAPI。 pip install fastapi uvicorn pip install langchain-community # 社区维护的各种工具和集成安装完成后不要急着跑。先做一个最小验证确保关键库能正常导入并且你的OpenAI API密钥或其他模型的密钥已正确设置环境变量。import os from langchain_openai import ChatOpenAI # 请在环境变量中设置你的OPENAI_API_KEY # export OPENAI_API_KEYyour-key-here (Linux/Mac) # set OPENAI_API_KEYyour-key-here (Windows) llm ChatOpenAI(modelgpt-4o-mini, temperature0) response llm.invoke(Hello, world!) print(response.content)如果能正常输出回复恭喜你基础环境通关。这里有个关键细节temperature参数控制生成文本的随机性。在Agent的“思考”环节我们通常设为较低值如0-0.2让它的决策更确定、可预测而在创意生成环节可以调高。这个参数会贯穿我们整个项目。2.2 开发工具选型VSCode与项目结构代码编辑器我选VSCode因为它对Python和Jupyter Notebook的支持非常出色。你需要安装Python扩展和Pylance。项目结构从一开始就要规划好避免后期变成一锅粥。ai_agent_project/ ├── app/ # FastAPI应用核心 │ ├── __init__.py │ ├── main.py # FastAPI应用入口 │ ├── agents/ # 存放不同功能的Agent │ │ ├── __init__.py │ │ └── research_agent.py │ └── tools/ # Agent可用的工具函数 │ ├── __init__.py │ └── web_search.py ├── knowledge_base/ # RAG知识库相关 │ ├── __init__.py │ ├── loader.py # 文档加载 │ ├── splitter.py # 文本分割 │ └── vector_store.py # 向量数据库操作 ├── graphs/ # LangGraph的状态图定义 │ ├── __init__.py │ └── sequential_agent.py ├── config.py # 配置文件API密钥等 ├── requirements.txt # 项目依赖 └── tests/ # 测试文件这样的结构清晰地将逻辑分层agents负责定义智能体的行为tools是它能调用的“手和脚”knowledge_base是它的“长期记忆”graphs则定义了它的“工作流程”。config.py用来集中管理配置千万不要把API密钥硬编码在代码里# config.py 示例 import os from dotenv import load_dotenv load_dotenv() # 从 .env 文件加载环境变量 class Config: OPENAI_API_KEY os.getenv(OPENAI_API_KEY) # 可以添加其他配置如向量数据库路径、模型名称等注意.env文件务必添加到.gitignore中防止密钥泄露。这是安全开发的第一课。3. RAG实战为AI构建一个“外部知识库”RAG检索增强生成是让AI突破其训练数据时间限制和私有数据访问壁垒的关键技术。它的原理不复杂把文档切块转换成向量一种数学表示存进向量数据库。当用户提问时将问题也转换成向量去数据库里找出最相似的几个文档块把这些“证据”和问题一起交给大模型让它生成基于这些证据的答案。3.1 文档加载与处理从PDF到文本块假设我们想给Agent装备一个关于“星际旅行指南”的私人知识库。我们有一些PDF和网页资料。第一步是加载它们。# knowledge_base/loader.py from langchain_community.document_loaders import PyPDFLoader, WebBaseLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from typing import List from langchain.schema import Document def load_documents(source_type: str, source_path: str) - List[Document]: 加载文档支持PDF和网页 documents [] if source_type pdf: loader PyPDFLoader(source_path) documents loader.load() elif source_type web: loader WebBaseLoader(source_path) documents loader.load() else: raise ValueError(fUnsupported source type: {source_type}) return documents def split_documents(documents: List[Document], chunk_size1000, chunk_overlap200) - List[Document]: 将长文档分割成适合嵌入的小块 text_splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, length_functionlen, separators[\n\n, \n, 。, , , , , , ] ) return text_splitter.split_documents(documents)这里有几个关键参数chunk_size每个文本块的最大字符数。太小会丢失上下文太大会降低检索精度并增加模型处理负担。1000是个不错的起点。chunk_overlap块之间的重叠字符数。这能防止一个完整的句子或概念被硬生生切断保留上下文连贯性。通常设为chunk_size的10%-20%。separators分割符列表按优先级尝试分割。这里配置了从中英文标点到空格的顺序能较好地处理混合文本。3.2 向量化与存储把文本变成可搜索的“地图”文本块准备好后需要把它们转换成向量一组数字这个过程叫“嵌入”。我们使用OpenAI的嵌入模型并将结果存入ChromaDB。# knowledge_base/vector_store.py import chromadb from chromadb.config import Settings from langchain_openai import OpenAIEmbeddings from langchain.vectorstores import Chroma import os from .splitter import split_documents class KnowledgeBase: def __init__(self, persist_directory./chroma_db): # 初始化嵌入模型 self.embeddings OpenAIEmbeddings(modeltext-embedding-3-small) # 初始化Chroma客户端设置持久化路径 self.client chromadb.PersistentClient(pathpersist_directory) # LangChain的Chroma包装器方便使用 self.vector_store None def create_from_documents(self, documents, collection_nameknowledge_base): 从文档创建或更新向量库 # 分割文档 splits split_documents(documents) # 创建集合并添加文档 self.vector_store Chroma.from_documents( documentssplits, embeddingself.embeddings, clientself.client, collection_namecollection_name, ) print(f知识库创建成功共存入 {len(splits)} 个文本块。) def query(self, question: str, k4) - List[Document]: 检索与问题最相关的k个文档块 if self.vector_store is None: raise ValueError(请先创建或加载知识库。) return self.vector_store.similarity_search(question, kk) # 使用示例 if __name__ __main__: kb KnowledgeBase() # 假设我们已经加载了documents # docs load_documents(pdf, ./data/starship_manual.pdf) # kb.create_from_documents(docs) # 查询 # results kb.query(曲速引擎的原理是什么) # for doc in results: # print(doc.page_content[:200], \n---\n)实操心得嵌入模型的选择直接影响检索质量。text-embedding-3-small在成本和性能间取得了很好的平衡。对于中文场景可以尝试text-embedding-3-small或专门的多语言模型。另外k值返回的文档块数量需要调优太小可能证据不足太大会引入噪声并增加API调用成本。可以从3开始根据回答质量调整。3.3 RAG链的组装从检索到生成有了知识库我们需要把它和语言模型连接起来形成一个完整的“提问-检索-回答”管道。# 这是一个简单的RAG链示例 from langchain.chains import RetrievalQA from langchain_openai import ChatOpenAI def create_rag_chain(vector_store): llm ChatOpenAI(modelgpt-4o-mini, temperature0) qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, # 最简单的方式将所有检索到的文档“塞”进提示词 retrievervector_store.as_retriever(search_kwargs{k: 4}), return_source_documentsTrue, # 返回源文档便于调试 verboseTrue, # 打印详细日志学习时很有用 ) return qa_chain # 使用链 # qa_chain create_rag_chain(kb.vector_store) # result qa_chain.invoke({query: 曲速引擎的原理是什么}) # print(result[result]) # print(来源文档, result[source_documents])这里的chain_typestuff是最直接的方法但它有上下文长度限制。如果检索到的文档总长度超过模型限制就需要考虑map_reduce、refine等其他更复杂但能处理长文档的链类型。对于初学者stuff在文档块不大时完全够用。4. 打造你的第一个原生Agent让AI学会使用工具RAG让AI有了记忆而Agent让AI有了“手”。一个原生Agent的核心是根据用户输入决定是直接回答还是调用某个工具比如计算器、搜索API、数据库然后把工具的结果整合起来最终给出回答。LangChain提供了AgentExecutor来简化这个过程。4.1 定义工具Agent的“技能包”工具本质上是一个函数有明确的名称、描述和参数。模型通过描述来决定是否以及如何调用它。# tools/calculator.py from langchain.tools import tool import math tool def calculator(expression: str) - str: 计算一个数学表达式的值。支持加减乘除(-*/)、乘方(**)、括号和常见函数如sqrt, sin, cos。表达式应为字符串。 try: # 安全评估使用math模块和限制内置函数避免eval的安全风险 # 这里为简化使用eval生产环境应用更安全的替代方案如ast.literal_eval或自定义解析器 allowed_names {k: v for k, v in math.__dict__.items() if not k.startswith(_)} allowed_names.update({abs: abs}) result eval(expression, {__builtins__: {}}, allowed_names) return str(result) except Exception as e: return f计算错误{e} # tools/web_search.py (模拟真实情况需接入SerpAPI等) from langchain.tools import tool import requests tool def search_web(query: str) - str: 在互联网上搜索最新信息。输入一个搜索查询词。 # 此处为模拟。真实集成时替换为SerpAPI或SearxNG等服务的调用 print(f[模拟搜索] 搜索词{query}) # 模拟返回一些结果 mock_results f 关于 {query} 的模拟搜索结果 1. 相关文章A介绍了{query}的基本概念。 2. 新闻B最近关于{query}的进展。 3. 教程C如何上手{query}。 return mock_results关键点工具的描述docstring至关重要大模型完全依赖这段描述来理解工具的用途和输入格式。描述要清晰、具体包含示例更好。tool装饰器会自动将函数转换成LangChain能识别的工具对象。4.2 创建Agent绑定大脑与工具我们将使用OpenAI的函数调用Function Calling能力来创建Agent这是目前最稳定和高效的方式之一。# agents/basic_agent.py from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from tools.calculator import calculator from tools.web_search import search_web def create_basic_agent(): # 1. 定义工具列表 tools [calculator, search_web] # 2. 定义提示词模板 prompt ChatPromptTemplate.from_messages([ (system, 你是一个乐于助人的助手。你可以使用工具来获取信息或进行计算。请清晰思考每一步。如果使用工具请提供完整的输入。), MessagesPlaceholder(variable_namechat_history), # 预留位置存放对话历史实现多轮对话 (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), # Agent的思考过程 ]) # 3. 选择大模型 llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 4. 创建Agent agent create_openai_tools_agent(llm, tools, prompt) # 5. 创建执行器 agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) return agent_executor # 运行Agent if __name__ __main__: agent create_basic_agent() result agent.invoke({input: 先计算一下(3的平方加上4的平方)再开根号是多少然后搜索一下关于勾股定理的最新文章。}) print(result[output])运行这段代码你会看到verboseTrue模式下打印的详细思考过程思考用户的问题需要先计算再搜索。我有一个计算器工具。行动调用calculator工具输入(3**2 4**2)**0.5。观察工具返回结果5.0。思考计算完成现在需要搜索。行动调用search_web工具输入勾股定理 最新文章。观察工具返回模拟的搜索结果。最终回答将计算和搜索的结果整合给出最终回复。这就是一个原生Agent的完整工作流思考-行动-观察-再思考...直到得出最终答案。AgentExecutor负责管理这个循环。4.3 处理复杂对话为Agent添加记忆上面的Agent是“单轮”的它不记得之前说过什么。要实现真正的对话需要引入记忆。LangChain提供了多种记忆后端最简单的是ConversationBufferMemory。from langchain.memory import ConversationBufferMemory def create_agent_with_memory(): tools [calculator, search_web] llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 关键创建记忆对象 memory ConversationBufferMemory(memory_keychat_history, return_messagesTrue) prompt ChatPromptTemplate.from_messages([ (system, 你是一个友好的助手。请利用之前的对话历史来更好地理解当前问题。), MessagesPlaceholder(variable_namechat_history), (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ]) agent create_openai_tools_agent(llm, tools, prompt) # 将memory传入executor agent_executor AgentExecutor( agentagent, toolstools, memorymemory, verboseTrue, handle_parsing_errorsTrue ) return agent_executor # 测试多轮对话 agent create_agent_with_memory() print(agent.invoke({input: 我叫小明})) # 输出: “你好小明” print(agent.invoke({input: 我刚才说我叫什么名字})) # 输出: “你刚才说你叫小明。”现在Agent有了对话记忆。ConversationBufferMemory简单地将所有历史对话保存在内存中。对于更复杂的场景你可能需要ConversationSummaryMemory总结历史或ConversationKGMemory知识图谱记忆。5. 引入LangGraph构建有状态、可循环的复杂Agent当任务变得复杂需要多个步骤、条件分支或循环时基础的AgentExecutor就显得力不从心了。比如你想让Agent“研究一个主题先搜索资料然后总结再根据总结提出三个问题最后回答问题”。这种带循环和状态的任务就需要LangGraph。5.1 理解LangGraph的核心状态State与节点NodeLangGraph把工作流建模成一个有向图。图的节点是函数或工具调用边定义了节点之间的流转条件。最关键的是整个图共享一个状态State这是一个字典随着流程推进不断更新。我们来构建一个“研究助手”Agent它包含两个主要节点一个“搜索”节点和一个“总结”节点并且可以循环。# graphs/research_agent.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage from tools.web_search import search_web # 假设我们有一个真实的搜索工具 # 1. 定义状态结构 class ResearchState(TypedDict): topic: str # 研究主题 search_results: List[str] # 搜索结果列表 summary: str # 总结内容 questions: List[str] # 生成的问题列表 answers: List[str] # 对问题的回答 iterations: Annotated[int, operator.add] # 循环次数每次1 # 2. 定义各个节点函数 def search_node(state: ResearchState) - ResearchState: 执行搜索的节点 print(f[搜索节点] 正在搜索主题: {state[topic]}) # 调用搜索工具 results search_web(state[topic]) # 更新状态 state[search_results].append(results) return state def summarize_node(state: ResearchState) - ResearchState: 总结搜索结果的节点 print(f[总结节点] 正在总结 {len(state[search_results])} 条结果...) llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 将最近的搜索结果拼接到一起 recent_results state[search_results][-1] # 取最后一次搜索结果 prompt f 请根据以下搜索内容为主题“{state[topic]}”撰写一份简洁的总结不超过200字。 搜索内容 {recent_results} response llm.invoke(prompt) state[summary] response.content return state def generate_questions_node(state: ResearchState) - ResearchState: 基于总结生成问题的节点 print([生成问题节点] 基于总结生成深入问题...) llm ChatOpenAI(modelgpt-4o-mini, temperature0.3) # 温度稍高鼓励创意 prompt f 基于以下总结提出三个能引发深入思考的问题 总结{state[summary]} 请直接输出三个问题每个问题占一行。 response llm.invoke(prompt) questions [q.strip() for q in response.content.split(\n) if q.strip()] state[questions] questions[:3] # 只取前三个 return state def should_continue(state: ResearchState) - str: 条件判断边决定是继续循环还是结束 # 简单逻辑如果循环次数少于2次且总结还不够长就继续搜索总结 if state[iterations] 2 and len(state.get(summary, )) 150: return search # 返回“search”边的名称继续循环 else: return END # 结束 # 3. 构建图 def create_research_graph(): workflow StateGraph(ResearchState) # 添加节点 workflow.add_node(search, search_node) workflow.add_node(summarize, summarize_node) workflow.add_node(generate_questions, generate_questions_node) # 设置入口点 workflow.set_entry_point(search) # 添加边定义流程 workflow.add_edge(search, summarize) workflow.add_edge(summarize, generate_questions) # 添加条件边从 generate_questions 出来后根据条件决定下一步 workflow.add_conditional_edges( generate_questions, should_continue, # 条件判断函数 { search: search, # 如果返回search则跳回search节点 END: END # 如果返回END则结束 } ) # 编译图 graph workflow.compile() return graph # 4. 运行图 if __name__ __main__: graph create_research_graph() # 初始化状态 initial_state { topic: 可控核聚变的最新进展, search_results: [], summary: , questions: [], answers: [], iterations: 0 } # 运行图 final_state graph.invoke(initial_state) print(\n 研究完成 ) print(f主题: {final_state[topic]}) print(f循环次数: {final_state[iterations]}) print(f总结: {final_state[summary][:300]}...) print(f生成的问题: {final_state[questions]})这个例子展示了LangGraph的强大之处状态管理所有节点都读取和更新同一个ResearchState字典。iterations字段使用了Annotated[int, operator.add]这是一个特殊注解告诉LangGraph在每次更新时对这个字段执行加法操作非常适合计数器。条件循环add_conditional_edges方法实现了条件分支。should_continue函数根据当前状态如总结长度、循环次数决定是返回search继续搜索还是END结束。这就形成了一个“搜索-总结-生成问题-判断是否继续搜索”的循环工作流。可视化LangGraph还有一个很棒的功能是可视化。你可以用graph.get_graph().draw_mermaid_png()需要安装pygraphviz来生成流程图直观看到你的Agent工作流。5.2 将RAG整合进LangGraph拥有长期记忆的Agent现在我们把前面构建的RAG知识库作为Agent的一个“工具节点”整合到LangGraph图中让Agent不仅能搜索网络还能查询自己的私有知识库。# graphs/agent_with_rag.py from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from knowledge_base.vector_store import KnowledgeBase # 导入之前写的知识库类 class AgentState(TypedDict): question: str knowledge: str # 从RAG知识库检索到的内容 reasoning: str # Agent的思考过程 final_answer: str steps: Annotated[list, operator.add] # 记录步骤 def create_rag_agent_graph(kb: KnowledgeBase): workflow StateGraph(AgentState) def retrieve_from_kb(state: AgentState): 节点从知识库检索 print(f[检索节点] 正在知识库中查询: {state[question]}) docs kb.query(state[question], k3) retrieved_text \n\n.join([doc.page_content for doc in docs]) state[knowledge] retrieved_text state[steps].append(f从知识库检索到 {len(docs)} 条相关信息。) return state def reason_and_answer(state: AgentState): 节点基于检索到的知识进行推理并回答 print([推理节点] 综合信息生成答案...) llm ChatOpenAI(modelgpt-4o-mini, temperature0) prompt f 你是一个专业助手。请基于以下背景知识回答用户的问题。 如果知识不足以回答问题请如实说明不要编造。 背景知识 {state[knowledge]} 用户问题 {state[question]} 请一步步推理然后给出最终答案。 response llm.invoke(prompt) # 我们可以尝试让模型分离“推理过程”和“最终答案” # 这里简单处理将全部内容作为答案 state[final_answer] response.content state[steps].append(综合检索信息生成最终答案。) return state def route_question(state: AgentState) - str: 路由节点决定是否需要查询知识库 # 这里可以加入更复杂的逻辑比如判断问题类型 # 例如如果问题包含特定关键词或属于某个领域就走知识库路径 # 这里简单假设所有问题都需要查知识库 if state.get(knowledge): # 如果已经有知识了直接去回答 return answer else: # 否则先去检索 return retrieve # 添加节点 workflow.add_node(retrieve, retrieve_from_kb) workflow.add_node(answer, reason_and_answer) # 设置条件路由 workflow.add_conditional_edges( start, # 假设我们有一个虚拟的起始节点实际可以用set_entry_point route_question, { retrieve: retrieve, answer: answer } ) workflow.add_edge(retrieve, answer) workflow.add_edge(answer, END) workflow.set_entry_point(start) return workflow.compile() # 使用 # kb KnowledgeBase(persist_directory./my_kb) # graph create_rag_agent_graph(kb) # result graph.invoke({question: 曲速引擎的原理是什么, knowledge: , reasoning: , final_answer: , steps: []})这个图实现了一个简单的决策流根据问题先路由到“检索”节点查询知识库然后将检索结果传递给“回答”节点生成最终答案。你可以扩展route_question函数实现更智能的路由比如简单计算类问题直接调用计算器工具事实性问题查询知识库最新信息查询网络搜索。6. 用FastAPI为Agent打造一个Web API到目前为止我们的Agent都在命令行里运行。要让它成为一个可被其他系统调用的服务我们需要一个API。FastAPI是一个现代、高性能的Python Web框架非常适合快速构建API。6.1 构建基础的Agent API端点我们将创建一个FastAPI应用提供两个端点一个用于问答一个用于管理知识库。# app/main.py from fastapi import FastAPI, HTTPException, UploadFile, File from pydantic import BaseModel from typing import List, Optional import os from agents.basic_agent import create_basic_agent from knowledge_base.vector_store import KnowledgeBase from graphs.agent_with_rag import create_rag_agent_graph app FastAPI(titleAI Agent API, description一个集成了RAG和LangGraph的智能体API) # 全局变量生产环境建议用更优雅的方式管理如依赖注入 agent_executor None rag_agent_graph None kb None class QueryRequest(BaseModel): question: str use_rag: bool True # 是否使用RAG知识库 conversation_id: Optional[str] None # 支持多会话 class QueryResponse(BaseModel): answer: str sources: List[str] [] # 答案来源如文档ID steps: List[str] [] # Agent执行的步骤 class KnowledgeUploadRequest(BaseModel): text: Optional[str] None # 直接上传文本 app.on_event(startup) async def startup_event(): 应用启动时初始化Agent和知识库 global agent_executor, kb, rag_agent_graph print(初始化AI Agent...) agent_executor create_basic_agent() # 加载或初始化知识库 kb KnowledgeBase(persist_directory./data/chroma_db) # 如果存在持久化数据可以在这里加载 # if os.path.exists(./data/chroma_db/chroma.sqlite3): # kb.load_collection(my_collection) # 创建RAG Agent图 rag_agent_graph create_rag_agent_graph(kb) print(启动完成。) app.post(/query, response_modelQueryResponse) async def query_agent(request: QueryRequest): 向Agent提问 if not agent_executor: raise HTTPException(status_code503, detailAgent未初始化) try: if request.use_rag and rag_agent_graph: # 使用带RAG的LangGraph Agent state rag_agent_graph.invoke({ question: request.question, knowledge: , reasoning: , final_answer: , steps: [] }) answer state.get(final_answer, 未能生成答案。) steps state.get(steps, []) # 这里可以解析出具体的来源文档ID sources [] else: # 使用基础工具调用Agent result agent_executor.invoke({input: request.question}) answer result.get(output, ) steps [f工具调用: {action} for action in result.get(intermediate_steps, [])] sources [] return QueryResponse(answeranswer, sourcessources, stepssteps) except Exception as e: raise HTTPException(status_code500, detailf处理查询时出错: {str(e)}) app.post(/knowledge/upload) async def upload_knowledge(file: UploadFile File(None), request: KnowledgeUploadRequest None): 上传文档到知识库支持文件或直接文本 if not kb: raise HTTPException(status_code503, detail知识库未初始化) documents [] if file: # 处理上传的文件示例仅支持文本文件 if file.content_type not in [text/plain, application/pdf]: raise HTTPException(status_code400, detail不支持的文件格式) contents await file.read() # 这里需要根据文件类型调用不同的loader例如PyPDFLoader # 为简化我们假设是文本 from langchain.schema import Document documents [Document(page_contentcontents.decode(utf-8), metadata{source: file.filename})] elif request and request.text: from langchain.schema import Document documents [Document(page_contentrequest.text, metadata{source: direct_input})] else: raise HTTPException(status_code400, detail未提供内容) try: kb.create_from_documents(documents, collection_nameuser_uploads) return {message: f成功上传 {len(documents)} 个文档到知识库。} except Exception as e: raise HTTPException(status_code500, detailf上传失败: {str(e)}) app.get(/health) async def health_check(): 健康检查端点 return {status: healthy, agent_ready: agent_executor is not None} if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)这个API提供了两个核心功能/query接收用户问题可以选择使用基础Agent还是RAG增强的Agent来回答。/knowledge/upload允许用户上传文本或文件来扩充Agent的私有知识库。部署注意生产环境中你需要处理更复杂的文件解析PDF、Word、Markdown等、异步处理、错误处理、身份验证和限流。startup_event中的全局变量在多个工作进程下会有问题应考虑使用外部存储如Redis来共享Agent状态或使用更高级的进程管理。6.2 测试你的API保存代码为main.py在终端运行uvicorn app.main:app --reload --host 0.0.0.0 --port 8000访问http://localhost:8000/docs你会看到自动生成的交互式API文档Swagger UI。你可以直接在浏览器里测试/query和/knowledge/upload端点。7. 避坑指南与性能优化从能跑到跑得好在把这套系统投入实际使用前还有一些关键的坑要避开以及优化点要考虑。7.1 常见错误与调试技巧OpenAI API密钥错误或超限这是最常见的问题。确保环境变量OPENAI_API_KEY设置正确。如果遇到速率限制Rate Limit需要在代码中增加重试逻辑和退避策略。可以使用tenacity库。from tenacity import retry, stop_after_attempt, wait_exponential from openai import RateLimitError retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def call_llm_with_retry(llm, prompt): try: return llm.invoke(prompt) except RateLimitError: print(触发速率限制等待后重试...) raise # tenacity会捕获这个异常并重试工具调用解析失败Agent有时会生成不符合工具输入格式的内容。在创建AgentExecutor时设置handle_parsing_errorsTrue可以让Agent在解析失败时尝试修复。更稳妥的做法是在工具函数内部做好健壮的错误处理返回清晰的错误信息供Agent理解。LangGraph状态图编译错误检查状态类TypedDict的定义是否正确特别是Annotated字段的用法。确保所有节点函数都接受并返回完整的状态字典。使用graph.get_graph().draw_mermaid_png()可视化你的图检查节点和边是否正确连接。向量数据库检索不准块大小不合适调整chunk_size和chunk_overlap。对于技术文档可能需要较小的块如500和较大的重叠如100。嵌入模型不匹配确保查询时使用的嵌入模型与建库时相同。元数据过滤在检索时可以添加元数据过滤条件如文档类型、日期提高精度。ChromaDB的as_retriever(search_kwargs{k: 4, filter: {category: manual}})支持此功能。7.2 性能与成本优化缓存对于频繁且不变的查询如知识库检索相同问题引入缓存可以极大减少LLM调用和嵌入计算。LangChain内置了InMemoryCache、RedisCache等。from langchain.cache import InMemoryCache from langchain.globals import set_llm_cache set_llm_cache(InMemoryCache())异步调用如果你的Agent需要并行调用多个工具如同时搜索多个关键词使用异步可以大幅缩短响应时间。FastAPI天然支持异步确保你的工具函数和LLM调用也是异步的如使用langchain_openai的ChatOpenAI的ainvoke方法。模型选择不是所有任务都需要GPT-4。对于工具调用路由、意图分类等简单任务gpt-3.5-turbo甚至更小的模型可能就足够了成本会低很多。对于最终答案生成再使用更强大的模型。提示词优化精心设计的提示词System Message和Few-Shot示例能显著提升Agent的决策质量和工具调用准确率减少不必要的循环或错误调用从而节省token。把指令写清楚、写具体。RAG检索优化重排序Re-ranking初步检索出10个文档块后用一个更小的、专门用于重排序的模型对它们进行打分排序只将Top-K个最相关的块送给LLM。这能提升答案质量并减少上下文长度。混合搜索结合关键词搜索如BM25和向量搜索取长补短。ChromaDB支持混合检索。7.3 扩展方向你的Agent还能做什么至此你已经拥有了一个功能完整的AI Agent框架。你可以在此基础上无限扩展更多工具集成数据库查询、发送邮件、调用企业内部API、控制智能家居等。更复杂的图实现多Agent协作一个负责研究一个负责写作一个负责审核、带有审批流程的自动化任务等。前端界面用Gradio或Streamlit快速搭建一个聊天界面或者用Vue/React构建更复杂的管理后台。记忆持久化将对话历史、知识库索引存储到PostgreSQL或MongoDB中实现跨会话记忆。监控与评估记录每一次查询的输入、输出、使用的工具、token消耗和用户反馈用于分析和优化Agent表现。这15天的旅程我们从环境搭建开始经历了RAG知识库构建、原生Agent开发、LangGraph工作流编排最终封装成Web API。每一个环节都配有可运行的代码和背后的原理讲解。希望这份详实的指南能成为你探索AI Agent世界的坚实起点。记住最好的学习方式就是动手改造它加入一个新工具设计一个新图解决一个你自己的实际问题。 从零构建AI Agent：RAG与LangGraph实战指南
1. 从零到一为什么我们需要Agent、RAG与LangGraph如果你最近在AI应用开发圈子里待过大概率已经被“Agent”、“RAG”这些词刷屏了。但说实话很多教程要么上来就甩给你一堆代码要么就是大谈特谈概念看完之后感觉懂了一动手就懵。这就像有人告诉你“开车很简单踩油门就走”但没告诉你离合器、档位和交通规则。我花了半个月时间从零开始把原生Agent、RAG和LangGraph这三个核心概念揉在一起用代码实操了一遍。这个过程里踩的坑、绕的弯以及最终跑通一个能“思考”、能“查资料”、能“按计划行事”的智能体的快感我想完整地分享给你。简单来说我们想解决的问题是让AI模型比如GPT从一个“知识截止到2023年初、只会被动回答问题”的聊天机器人进化成一个能主动使用工具、访问最新或私有知识、并按照复杂逻辑执行多步骤任务的智能助手。这就是Agent智能体的愿景。而实现它离不开两样东西RAG检索增强生成给AI装上“外部记忆库”LangGraph则提供了编排这些复杂任务的“流程图”和“状态机”。接下来这15天的内容我会带你一步步搭建环境理解核心原理并用完整的代码实现一个具备长期记忆、能调用工具、可规划任务链的AI Agent。无论你是刚学完Python基础想找项目练手还是已经有一定经验想深入AI应用层开发这个系列都能给你一条清晰的路径。2. 环境搭建与核心库全景图不只是pip install工欲善其事必先利其器。在开始写第一行Agent代码之前一个稳定、可复现的开发环境至关重要。很多人在这里就会遇到第一个拦路虎。2.1 Python与包管理避开版本地狱首先确保你安装了Python 3.10或以上版本。这是大多数现代AI库的基线要求。我强烈建议使用conda或venv创建独立的虚拟环境这是避免包冲突的生命线。# 使用conda推荐尤其对Windows用户友好 conda create -n ai-agent python3.10 conda activate ai-agent # 或者使用venv python -m venv ai-agent # Windows激活 ai-agent\Scripts\activate # Mac/Linux激活 source ai-agent/bin/activate接下来是安装核心库。别急着一次性pip install所有东西我们分层进行。# 1. 基础框架层LangChain和LangGraph # LangChain是构建AI应用链式的工具箱LangGraph是它的扩展用于构建有状态的、循环的图即Agent的核心 pip install langchain langgraph # 2. 大模型接口层我们需要一个“大脑”。这里以OpenAI为例你也可以用Anthropic、Ollama本地模型等。 pip install openai # 3. 向量数据库与嵌入层这是RAG的“记忆库”部分。我们选用轻量且流行的ChromaDB。 pip install chromadb # 4. Web框架与工具层为了让Agent能对外提供服务或调用外部API我们引入FastAPI。 pip install fastapi uvicorn pip install langchain-community # 社区维护的各种工具和集成安装完成后不要急着跑。先做一个最小验证确保关键库能正常导入并且你的OpenAI API密钥或其他模型的密钥已正确设置环境变量。import os from langchain_openai import ChatOpenAI # 请在环境变量中设置你的OPENAI_API_KEY # export OPENAI_API_KEYyour-key-here (Linux/Mac) # set OPENAI_API_KEYyour-key-here (Windows) llm ChatOpenAI(modelgpt-4o-mini, temperature0) response llm.invoke(Hello, world!) print(response.content)如果能正常输出回复恭喜你基础环境通关。这里有个关键细节temperature参数控制生成文本的随机性。在Agent的“思考”环节我们通常设为较低值如0-0.2让它的决策更确定、可预测而在创意生成环节可以调高。这个参数会贯穿我们整个项目。2.2 开发工具选型VSCode与项目结构代码编辑器我选VSCode因为它对Python和Jupyter Notebook的支持非常出色。你需要安装Python扩展和Pylance。项目结构从一开始就要规划好避免后期变成一锅粥。ai_agent_project/ ├── app/ # FastAPI应用核心 │ ├── __init__.py │ ├── main.py # FastAPI应用入口 │ ├── agents/ # 存放不同功能的Agent │ │ ├── __init__.py │ │ └── research_agent.py │ └── tools/ # Agent可用的工具函数 │ ├── __init__.py │ └── web_search.py ├── knowledge_base/ # RAG知识库相关 │ ├── __init__.py │ ├── loader.py # 文档加载 │ ├── splitter.py # 文本分割 │ └── vector_store.py # 向量数据库操作 ├── graphs/ # LangGraph的状态图定义 │ ├── __init__.py │ └── sequential_agent.py ├── config.py # 配置文件API密钥等 ├── requirements.txt # 项目依赖 └── tests/ # 测试文件这样的结构清晰地将逻辑分层agents负责定义智能体的行为tools是它能调用的“手和脚”knowledge_base是它的“长期记忆”graphs则定义了它的“工作流程”。config.py用来集中管理配置千万不要把API密钥硬编码在代码里# config.py 示例 import os from dotenv import load_dotenv load_dotenv() # 从 .env 文件加载环境变量 class Config: OPENAI_API_KEY os.getenv(OPENAI_API_KEY) # 可以添加其他配置如向量数据库路径、模型名称等注意.env文件务必添加到.gitignore中防止密钥泄露。这是安全开发的第一课。3. RAG实战为AI构建一个“外部知识库”RAG检索增强生成是让AI突破其训练数据时间限制和私有数据访问壁垒的关键技术。它的原理不复杂把文档切块转换成向量一种数学表示存进向量数据库。当用户提问时将问题也转换成向量去数据库里找出最相似的几个文档块把这些“证据”和问题一起交给大模型让它生成基于这些证据的答案。3.1 文档加载与处理从PDF到文本块假设我们想给Agent装备一个关于“星际旅行指南”的私人知识库。我们有一些PDF和网页资料。第一步是加载它们。# knowledge_base/loader.py from langchain_community.document_loaders import PyPDFLoader, WebBaseLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from typing import List from langchain.schema import Document def load_documents(source_type: str, source_path: str) - List[Document]: 加载文档支持PDF和网页 documents [] if source_type pdf: loader PyPDFLoader(source_path) documents loader.load() elif source_type web: loader WebBaseLoader(source_path) documents loader.load() else: raise ValueError(fUnsupported source type: {source_type}) return documents def split_documents(documents: List[Document], chunk_size1000, chunk_overlap200) - List[Document]: 将长文档分割成适合嵌入的小块 text_splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, length_functionlen, separators[\n\n, \n, 。, , , , , , ] ) return text_splitter.split_documents(documents)这里有几个关键参数chunk_size每个文本块的最大字符数。太小会丢失上下文太大会降低检索精度并增加模型处理负担。1000是个不错的起点。chunk_overlap块之间的重叠字符数。这能防止一个完整的句子或概念被硬生生切断保留上下文连贯性。通常设为chunk_size的10%-20%。separators分割符列表按优先级尝试分割。这里配置了从中英文标点到空格的顺序能较好地处理混合文本。3.2 向量化与存储把文本变成可搜索的“地图”文本块准备好后需要把它们转换成向量一组数字这个过程叫“嵌入”。我们使用OpenAI的嵌入模型并将结果存入ChromaDB。# knowledge_base/vector_store.py import chromadb from chromadb.config import Settings from langchain_openai import OpenAIEmbeddings from langchain.vectorstores import Chroma import os from .splitter import split_documents class KnowledgeBase: def __init__(self, persist_directory./chroma_db): # 初始化嵌入模型 self.embeddings OpenAIEmbeddings(modeltext-embedding-3-small) # 初始化Chroma客户端设置持久化路径 self.client chromadb.PersistentClient(pathpersist_directory) # LangChain的Chroma包装器方便使用 self.vector_store None def create_from_documents(self, documents, collection_nameknowledge_base): 从文档创建或更新向量库 # 分割文档 splits split_documents(documents) # 创建集合并添加文档 self.vector_store Chroma.from_documents( documentssplits, embeddingself.embeddings, clientself.client, collection_namecollection_name, ) print(f知识库创建成功共存入 {len(splits)} 个文本块。) def query(self, question: str, k4) - List[Document]: 检索与问题最相关的k个文档块 if self.vector_store is None: raise ValueError(请先创建或加载知识库。) return self.vector_store.similarity_search(question, kk) # 使用示例 if __name__ __main__: kb KnowledgeBase() # 假设我们已经加载了documents # docs load_documents(pdf, ./data/starship_manual.pdf) # kb.create_from_documents(docs) # 查询 # results kb.query(曲速引擎的原理是什么) # for doc in results: # print(doc.page_content[:200], \n---\n)实操心得嵌入模型的选择直接影响检索质量。text-embedding-3-small在成本和性能间取得了很好的平衡。对于中文场景可以尝试text-embedding-3-small或专门的多语言模型。另外k值返回的文档块数量需要调优太小可能证据不足太大会引入噪声并增加API调用成本。可以从3开始根据回答质量调整。3.3 RAG链的组装从检索到生成有了知识库我们需要把它和语言模型连接起来形成一个完整的“提问-检索-回答”管道。# 这是一个简单的RAG链示例 from langchain.chains import RetrievalQA from langchain_openai import ChatOpenAI def create_rag_chain(vector_store): llm ChatOpenAI(modelgpt-4o-mini, temperature0) qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, # 最简单的方式将所有检索到的文档“塞”进提示词 retrievervector_store.as_retriever(search_kwargs{k: 4}), return_source_documentsTrue, # 返回源文档便于调试 verboseTrue, # 打印详细日志学习时很有用 ) return qa_chain # 使用链 # qa_chain create_rag_chain(kb.vector_store) # result qa_chain.invoke({query: 曲速引擎的原理是什么}) # print(result[result]) # print(来源文档, result[source_documents])这里的chain_typestuff是最直接的方法但它有上下文长度限制。如果检索到的文档总长度超过模型限制就需要考虑map_reduce、refine等其他更复杂但能处理长文档的链类型。对于初学者stuff在文档块不大时完全够用。4. 打造你的第一个原生Agent让AI学会使用工具RAG让AI有了记忆而Agent让AI有了“手”。一个原生Agent的核心是根据用户输入决定是直接回答还是调用某个工具比如计算器、搜索API、数据库然后把工具的结果整合起来最终给出回答。LangChain提供了AgentExecutor来简化这个过程。4.1 定义工具Agent的“技能包”工具本质上是一个函数有明确的名称、描述和参数。模型通过描述来决定是否以及如何调用它。# tools/calculator.py from langchain.tools import tool import math tool def calculator(expression: str) - str: 计算一个数学表达式的值。支持加减乘除(-*/)、乘方(**)、括号和常见函数如sqrt, sin, cos。表达式应为字符串。 try: # 安全评估使用math模块和限制内置函数避免eval的安全风险 # 这里为简化使用eval生产环境应用更安全的替代方案如ast.literal_eval或自定义解析器 allowed_names {k: v for k, v in math.__dict__.items() if not k.startswith(_)} allowed_names.update({abs: abs}) result eval(expression, {__builtins__: {}}, allowed_names) return str(result) except Exception as e: return f计算错误{e} # tools/web_search.py (模拟真实情况需接入SerpAPI等) from langchain.tools import tool import requests tool def search_web(query: str) - str: 在互联网上搜索最新信息。输入一个搜索查询词。 # 此处为模拟。真实集成时替换为SerpAPI或SearxNG等服务的调用 print(f[模拟搜索] 搜索词{query}) # 模拟返回一些结果 mock_results f 关于 {query} 的模拟搜索结果 1. 相关文章A介绍了{query}的基本概念。 2. 新闻B最近关于{query}的进展。 3. 教程C如何上手{query}。 return mock_results关键点工具的描述docstring至关重要大模型完全依赖这段描述来理解工具的用途和输入格式。描述要清晰、具体包含示例更好。tool装饰器会自动将函数转换成LangChain能识别的工具对象。4.2 创建Agent绑定大脑与工具我们将使用OpenAI的函数调用Function Calling能力来创建Agent这是目前最稳定和高效的方式之一。# agents/basic_agent.py from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from tools.calculator import calculator from tools.web_search import search_web def create_basic_agent(): # 1. 定义工具列表 tools [calculator, search_web] # 2. 定义提示词模板 prompt ChatPromptTemplate.from_messages([ (system, 你是一个乐于助人的助手。你可以使用工具来获取信息或进行计算。请清晰思考每一步。如果使用工具请提供完整的输入。), MessagesPlaceholder(variable_namechat_history), # 预留位置存放对话历史实现多轮对话 (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), # Agent的思考过程 ]) # 3. 选择大模型 llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 4. 创建Agent agent create_openai_tools_agent(llm, tools, prompt) # 5. 创建执行器 agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) return agent_executor # 运行Agent if __name__ __main__: agent create_basic_agent() result agent.invoke({input: 先计算一下(3的平方加上4的平方)再开根号是多少然后搜索一下关于勾股定理的最新文章。}) print(result[output])运行这段代码你会看到verboseTrue模式下打印的详细思考过程思考用户的问题需要先计算再搜索。我有一个计算器工具。行动调用calculator工具输入(3**2 4**2)**0.5。观察工具返回结果5.0。思考计算完成现在需要搜索。行动调用search_web工具输入勾股定理 最新文章。观察工具返回模拟的搜索结果。最终回答将计算和搜索的结果整合给出最终回复。这就是一个原生Agent的完整工作流思考-行动-观察-再思考...直到得出最终答案。AgentExecutor负责管理这个循环。4.3 处理复杂对话为Agent添加记忆上面的Agent是“单轮”的它不记得之前说过什么。要实现真正的对话需要引入记忆。LangChain提供了多种记忆后端最简单的是ConversationBufferMemory。from langchain.memory import ConversationBufferMemory def create_agent_with_memory(): tools [calculator, search_web] llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 关键创建记忆对象 memory ConversationBufferMemory(memory_keychat_history, return_messagesTrue) prompt ChatPromptTemplate.from_messages([ (system, 你是一个友好的助手。请利用之前的对话历史来更好地理解当前问题。), MessagesPlaceholder(variable_namechat_history), (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ]) agent create_openai_tools_agent(llm, tools, prompt) # 将memory传入executor agent_executor AgentExecutor( agentagent, toolstools, memorymemory, verboseTrue, handle_parsing_errorsTrue ) return agent_executor # 测试多轮对话 agent create_agent_with_memory() print(agent.invoke({input: 我叫小明})) # 输出: “你好小明” print(agent.invoke({input: 我刚才说我叫什么名字})) # 输出: “你刚才说你叫小明。”现在Agent有了对话记忆。ConversationBufferMemory简单地将所有历史对话保存在内存中。对于更复杂的场景你可能需要ConversationSummaryMemory总结历史或ConversationKGMemory知识图谱记忆。5. 引入LangGraph构建有状态、可循环的复杂Agent当任务变得复杂需要多个步骤、条件分支或循环时基础的AgentExecutor就显得力不从心了。比如你想让Agent“研究一个主题先搜索资料然后总结再根据总结提出三个问题最后回答问题”。这种带循环和状态的任务就需要LangGraph。5.1 理解LangGraph的核心状态State与节点NodeLangGraph把工作流建模成一个有向图。图的节点是函数或工具调用边定义了节点之间的流转条件。最关键的是整个图共享一个状态State这是一个字典随着流程推进不断更新。我们来构建一个“研究助手”Agent它包含两个主要节点一个“搜索”节点和一个“总结”节点并且可以循环。# graphs/research_agent.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage from tools.web_search import search_web # 假设我们有一个真实的搜索工具 # 1. 定义状态结构 class ResearchState(TypedDict): topic: str # 研究主题 search_results: List[str] # 搜索结果列表 summary: str # 总结内容 questions: List[str] # 生成的问题列表 answers: List[str] # 对问题的回答 iterations: Annotated[int, operator.add] # 循环次数每次1 # 2. 定义各个节点函数 def search_node(state: ResearchState) - ResearchState: 执行搜索的节点 print(f[搜索节点] 正在搜索主题: {state[topic]}) # 调用搜索工具 results search_web(state[topic]) # 更新状态 state[search_results].append(results) return state def summarize_node(state: ResearchState) - ResearchState: 总结搜索结果的节点 print(f[总结节点] 正在总结 {len(state[search_results])} 条结果...) llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 将最近的搜索结果拼接到一起 recent_results state[search_results][-1] # 取最后一次搜索结果 prompt f 请根据以下搜索内容为主题“{state[topic]}”撰写一份简洁的总结不超过200字。 搜索内容 {recent_results} response llm.invoke(prompt) state[summary] response.content return state def generate_questions_node(state: ResearchState) - ResearchState: 基于总结生成问题的节点 print([生成问题节点] 基于总结生成深入问题...) llm ChatOpenAI(modelgpt-4o-mini, temperature0.3) # 温度稍高鼓励创意 prompt f 基于以下总结提出三个能引发深入思考的问题 总结{state[summary]} 请直接输出三个问题每个问题占一行。 response llm.invoke(prompt) questions [q.strip() for q in response.content.split(\n) if q.strip()] state[questions] questions[:3] # 只取前三个 return state def should_continue(state: ResearchState) - str: 条件判断边决定是继续循环还是结束 # 简单逻辑如果循环次数少于2次且总结还不够长就继续搜索总结 if state[iterations] 2 and len(state.get(summary, )) 150: return search # 返回“search”边的名称继续循环 else: return END # 结束 # 3. 构建图 def create_research_graph(): workflow StateGraph(ResearchState) # 添加节点 workflow.add_node(search, search_node) workflow.add_node(summarize, summarize_node) workflow.add_node(generate_questions, generate_questions_node) # 设置入口点 workflow.set_entry_point(search) # 添加边定义流程 workflow.add_edge(search, summarize) workflow.add_edge(summarize, generate_questions) # 添加条件边从 generate_questions 出来后根据条件决定下一步 workflow.add_conditional_edges( generate_questions, should_continue, # 条件判断函数 { search: search, # 如果返回search则跳回search节点 END: END # 如果返回END则结束 } ) # 编译图 graph workflow.compile() return graph # 4. 运行图 if __name__ __main__: graph create_research_graph() # 初始化状态 initial_state { topic: 可控核聚变的最新进展, search_results: [], summary: , questions: [], answers: [], iterations: 0 } # 运行图 final_state graph.invoke(initial_state) print(\n 研究完成 ) print(f主题: {final_state[topic]}) print(f循环次数: {final_state[iterations]}) print(f总结: {final_state[summary][:300]}...) print(f生成的问题: {final_state[questions]})这个例子展示了LangGraph的强大之处状态管理所有节点都读取和更新同一个ResearchState字典。iterations字段使用了Annotated[int, operator.add]这是一个特殊注解告诉LangGraph在每次更新时对这个字段执行加法操作非常适合计数器。条件循环add_conditional_edges方法实现了条件分支。should_continue函数根据当前状态如总结长度、循环次数决定是返回search继续搜索还是END结束。这就形成了一个“搜索-总结-生成问题-判断是否继续搜索”的循环工作流。可视化LangGraph还有一个很棒的功能是可视化。你可以用graph.get_graph().draw_mermaid_png()需要安装pygraphviz来生成流程图直观看到你的Agent工作流。5.2 将RAG整合进LangGraph拥有长期记忆的Agent现在我们把前面构建的RAG知识库作为Agent的一个“工具节点”整合到LangGraph图中让Agent不仅能搜索网络还能查询自己的私有知识库。# graphs/agent_with_rag.py from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from knowledge_base.vector_store import KnowledgeBase # 导入之前写的知识库类 class AgentState(TypedDict): question: str knowledge: str # 从RAG知识库检索到的内容 reasoning: str # Agent的思考过程 final_answer: str steps: Annotated[list, operator.add] # 记录步骤 def create_rag_agent_graph(kb: KnowledgeBase): workflow StateGraph(AgentState) def retrieve_from_kb(state: AgentState): 节点从知识库检索 print(f[检索节点] 正在知识库中查询: {state[question]}) docs kb.query(state[question], k3) retrieved_text \n\n.join([doc.page_content for doc in docs]) state[knowledge] retrieved_text state[steps].append(f从知识库检索到 {len(docs)} 条相关信息。) return state def reason_and_answer(state: AgentState): 节点基于检索到的知识进行推理并回答 print([推理节点] 综合信息生成答案...) llm ChatOpenAI(modelgpt-4o-mini, temperature0) prompt f 你是一个专业助手。请基于以下背景知识回答用户的问题。 如果知识不足以回答问题请如实说明不要编造。 背景知识 {state[knowledge]} 用户问题 {state[question]} 请一步步推理然后给出最终答案。 response llm.invoke(prompt) # 我们可以尝试让模型分离“推理过程”和“最终答案” # 这里简单处理将全部内容作为答案 state[final_answer] response.content state[steps].append(综合检索信息生成最终答案。) return state def route_question(state: AgentState) - str: 路由节点决定是否需要查询知识库 # 这里可以加入更复杂的逻辑比如判断问题类型 # 例如如果问题包含特定关键词或属于某个领域就走知识库路径 # 这里简单假设所有问题都需要查知识库 if state.get(knowledge): # 如果已经有知识了直接去回答 return answer else: # 否则先去检索 return retrieve # 添加节点 workflow.add_node(retrieve, retrieve_from_kb) workflow.add_node(answer, reason_and_answer) # 设置条件路由 workflow.add_conditional_edges( start, # 假设我们有一个虚拟的起始节点实际可以用set_entry_point route_question, { retrieve: retrieve, answer: answer } ) workflow.add_edge(retrieve, answer) workflow.add_edge(answer, END) workflow.set_entry_point(start) return workflow.compile() # 使用 # kb KnowledgeBase(persist_directory./my_kb) # graph create_rag_agent_graph(kb) # result graph.invoke({question: 曲速引擎的原理是什么, knowledge: , reasoning: , final_answer: , steps: []})这个图实现了一个简单的决策流根据问题先路由到“检索”节点查询知识库然后将检索结果传递给“回答”节点生成最终答案。你可以扩展route_question函数实现更智能的路由比如简单计算类问题直接调用计算器工具事实性问题查询知识库最新信息查询网络搜索。6. 用FastAPI为Agent打造一个Web API到目前为止我们的Agent都在命令行里运行。要让它成为一个可被其他系统调用的服务我们需要一个API。FastAPI是一个现代、高性能的Python Web框架非常适合快速构建API。6.1 构建基础的Agent API端点我们将创建一个FastAPI应用提供两个端点一个用于问答一个用于管理知识库。# app/main.py from fastapi import FastAPI, HTTPException, UploadFile, File from pydantic import BaseModel from typing import List, Optional import os from agents.basic_agent import create_basic_agent from knowledge_base.vector_store import KnowledgeBase from graphs.agent_with_rag import create_rag_agent_graph app FastAPI(titleAI Agent API, description一个集成了RAG和LangGraph的智能体API) # 全局变量生产环境建议用更优雅的方式管理如依赖注入 agent_executor None rag_agent_graph None kb None class QueryRequest(BaseModel): question: str use_rag: bool True # 是否使用RAG知识库 conversation_id: Optional[str] None # 支持多会话 class QueryResponse(BaseModel): answer: str sources: List[str] [] # 答案来源如文档ID steps: List[str] [] # Agent执行的步骤 class KnowledgeUploadRequest(BaseModel): text: Optional[str] None # 直接上传文本 app.on_event(startup) async def startup_event(): 应用启动时初始化Agent和知识库 global agent_executor, kb, rag_agent_graph print(初始化AI Agent...) agent_executor create_basic_agent() # 加载或初始化知识库 kb KnowledgeBase(persist_directory./data/chroma_db) # 如果存在持久化数据可以在这里加载 # if os.path.exists(./data/chroma_db/chroma.sqlite3): # kb.load_collection(my_collection) # 创建RAG Agent图 rag_agent_graph create_rag_agent_graph(kb) print(启动完成。) app.post(/query, response_modelQueryResponse) async def query_agent(request: QueryRequest): 向Agent提问 if not agent_executor: raise HTTPException(status_code503, detailAgent未初始化) try: if request.use_rag and rag_agent_graph: # 使用带RAG的LangGraph Agent state rag_agent_graph.invoke({ question: request.question, knowledge: , reasoning: , final_answer: , steps: [] }) answer state.get(final_answer, 未能生成答案。) steps state.get(steps, []) # 这里可以解析出具体的来源文档ID sources [] else: # 使用基础工具调用Agent result agent_executor.invoke({input: request.question}) answer result.get(output, ) steps [f工具调用: {action} for action in result.get(intermediate_steps, [])] sources [] return QueryResponse(answeranswer, sourcessources, stepssteps) except Exception as e: raise HTTPException(status_code500, detailf处理查询时出错: {str(e)}) app.post(/knowledge/upload) async def upload_knowledge(file: UploadFile File(None), request: KnowledgeUploadRequest None): 上传文档到知识库支持文件或直接文本 if not kb: raise HTTPException(status_code503, detail知识库未初始化) documents [] if file: # 处理上传的文件示例仅支持文本文件 if file.content_type not in [text/plain, application/pdf]: raise HTTPException(status_code400, detail不支持的文件格式) contents await file.read() # 这里需要根据文件类型调用不同的loader例如PyPDFLoader # 为简化我们假设是文本 from langchain.schema import Document documents [Document(page_contentcontents.decode(utf-8), metadata{source: file.filename})] elif request and request.text: from langchain.schema import Document documents [Document(page_contentrequest.text, metadata{source: direct_input})] else: raise HTTPException(status_code400, detail未提供内容) try: kb.create_from_documents(documents, collection_nameuser_uploads) return {message: f成功上传 {len(documents)} 个文档到知识库。} except Exception as e: raise HTTPException(status_code500, detailf上传失败: {str(e)}) app.get(/health) async def health_check(): 健康检查端点 return {status: healthy, agent_ready: agent_executor is not None} if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)这个API提供了两个核心功能/query接收用户问题可以选择使用基础Agent还是RAG增强的Agent来回答。/knowledge/upload允许用户上传文本或文件来扩充Agent的私有知识库。部署注意生产环境中你需要处理更复杂的文件解析PDF、Word、Markdown等、异步处理、错误处理、身份验证和限流。startup_event中的全局变量在多个工作进程下会有问题应考虑使用外部存储如Redis来共享Agent状态或使用更高级的进程管理。6.2 测试你的API保存代码为main.py在终端运行uvicorn app.main:app --reload --host 0.0.0.0 --port 8000访问http://localhost:8000/docs你会看到自动生成的交互式API文档Swagger UI。你可以直接在浏览器里测试/query和/knowledge/upload端点。7. 避坑指南与性能优化从能跑到跑得好在把这套系统投入实际使用前还有一些关键的坑要避开以及优化点要考虑。7.1 常见错误与调试技巧OpenAI API密钥错误或超限这是最常见的问题。确保环境变量OPENAI_API_KEY设置正确。如果遇到速率限制Rate Limit需要在代码中增加重试逻辑和退避策略。可以使用tenacity库。from tenacity import retry, stop_after_attempt, wait_exponential from openai import RateLimitError retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def call_llm_with_retry(llm, prompt): try: return llm.invoke(prompt) except RateLimitError: print(触发速率限制等待后重试...) raise # tenacity会捕获这个异常并重试工具调用解析失败Agent有时会生成不符合工具输入格式的内容。在创建AgentExecutor时设置handle_parsing_errorsTrue可以让Agent在解析失败时尝试修复。更稳妥的做法是在工具函数内部做好健壮的错误处理返回清晰的错误信息供Agent理解。LangGraph状态图编译错误检查状态类TypedDict的定义是否正确特别是Annotated字段的用法。确保所有节点函数都接受并返回完整的状态字典。使用graph.get_graph().draw_mermaid_png()可视化你的图检查节点和边是否正确连接。向量数据库检索不准块大小不合适调整chunk_size和chunk_overlap。对于技术文档可能需要较小的块如500和较大的重叠如100。嵌入模型不匹配确保查询时使用的嵌入模型与建库时相同。元数据过滤在检索时可以添加元数据过滤条件如文档类型、日期提高精度。ChromaDB的as_retriever(search_kwargs{k: 4, filter: {category: manual}})支持此功能。7.2 性能与成本优化缓存对于频繁且不变的查询如知识库检索相同问题引入缓存可以极大减少LLM调用和嵌入计算。LangChain内置了InMemoryCache、RedisCache等。from langchain.cache import InMemoryCache from langchain.globals import set_llm_cache set_llm_cache(InMemoryCache())异步调用如果你的Agent需要并行调用多个工具如同时搜索多个关键词使用异步可以大幅缩短响应时间。FastAPI天然支持异步确保你的工具函数和LLM调用也是异步的如使用langchain_openai的ChatOpenAI的ainvoke方法。模型选择不是所有任务都需要GPT-4。对于工具调用路由、意图分类等简单任务gpt-3.5-turbo甚至更小的模型可能就足够了成本会低很多。对于最终答案生成再使用更强大的模型。提示词优化精心设计的提示词System Message和Few-Shot示例能显著提升Agent的决策质量和工具调用准确率减少不必要的循环或错误调用从而节省token。把指令写清楚、写具体。RAG检索优化重排序Re-ranking初步检索出10个文档块后用一个更小的、专门用于重排序的模型对它们进行打分排序只将Top-K个最相关的块送给LLM。这能提升答案质量并减少上下文长度。混合搜索结合关键词搜索如BM25和向量搜索取长补短。ChromaDB支持混合检索。7.3 扩展方向你的Agent还能做什么至此你已经拥有了一个功能完整的AI Agent框架。你可以在此基础上无限扩展更多工具集成数据库查询、发送邮件、调用企业内部API、控制智能家居等。更复杂的图实现多Agent协作一个负责研究一个负责写作一个负责审核、带有审批流程的自动化任务等。前端界面用Gradio或Streamlit快速搭建一个聊天界面或者用Vue/React构建更复杂的管理后台。记忆持久化将对话历史、知识库索引存储到PostgreSQL或MongoDB中实现跨会话记忆。监控与评估记录每一次查询的输入、输出、使用的工具、token消耗和用户反馈用于分析和优化Agent表现。这15天的旅程我们从环境搭建开始经历了RAG知识库构建、原生Agent开发、LangGraph工作流编排最终封装成Web API。每一个环节都配有可运行的代码和背后的原理讲解。希望这份详实的指南能成为你探索AI Agent世界的坚实起点。记住最好的学习方式就是动手改造它加入一个新工具设计一个新图解决一个你自己的实际问题。