这次我们来看一个关于提示词工程Prompt Engineering和现代大模型应用开发栈的综合性技术主题。标题中提到的“吴恩达”和“2026公认最好的”更多是吸引眼球的表述但其核心指向的是当前大模型应用开发中几个最关键的架构范式Prompt Engineering提示词工程、LangGraph、RAG检索增强生成和Agent智能体。这并非一个单一的“项目”而是一套构建复杂、可靠AI应用的方法论和工具链。对于开发者而言最关心的不是哪个概念最火而是这套技术栈能不能用、怎么用、门槛高不高、以及如何整合。本文将抛开营销术语直接切入技术核心带你快速理解这四大支柱并通过一个整合了LangGraph、RAG和Agent的实战代码示例演示如何构建一个具备记忆、工具调用和知识检索能力的智能体应用。我们将重点关注其架构设计、代码实现、运行方式以及在实际场景中的效果验证。1. 核心能力速览下表概括了本次探讨的技术栈核心组件及其在应用开发中的角色能力项说明技术栈构成Prompt Engineering LangGraph RAG Agent核心定位构建复杂、可控制、具备长期记忆和外部知识的大模型应用框架主要功能结构化提示设计、有状态的工作流编排、外部知识检索、自主工具调用与任务分解硬件门槛开发阶段普通CPU/GPU均可依赖云端大模型API如OpenAI、DeepSeek。本地部署如需本地运行大模型则需相应GPU资源。启动方式基于Python脚本启动可通过FastAPI等封装为Web API服务。是否支持API是。核心逻辑可轻松封装为RESTful或GraphQL API供前端或其他服务调用。是否支持批量/异步任务是。LangGraph的工作流和Agent任务可以设计为处理队列任务。适合场景智能客服、复杂问答系统、自动化研究助手、个性化内容生成、数据分析Agent等需要多步骤推理和知识结合的场景。简单来说这套组合拳解决了单纯调用大模型API的几大痛点提示词效果不稳定、无法执行多轮复杂任务、缺乏私有知识、以及无法操作外部工具。2. 适用场景与使用边界适合谁全栈/后端开发者希望将大模型能力深度集成到现有产品中。AI应用创业者需要快速构建具备复杂逻辑的AI原型或产品。技术团队寻求标准化、可维护的大模型应用开发框架。能解决什么问题可控的对话与任务流通过LangGraph将多轮对话、工具调用、条件分支编排成可视化的工作流避免代码 spaghetti。知识库问答通过RAG让大模型能够回答关于特定领域、私有文档如产品手册、内部Wiki的问题避免“幻觉”。自动化智能体通过Agent框架让AI能够理解用户目标自主规划步骤、调用搜索引擎、计算器、数据库等工具完成任务。提示词标准化通过Prompt Engineering最佳实践将有效的提示词模板化、参数化提升生成效果的一致性和可复用性。不适合什么场景极其简单的单次问答如果只是调用ChatCompletion一次就能解决引入全套框架反而过度设计。对延迟极其敏感的场景RAG检索、多步Agent推理会增加响应时间。完全离线的环境如果无法连接任何大模型API包括本地部署的模型则无法运行。安全与合规边界知识版权RAG使用的文档需确保拥有合法使用权避免侵犯版权。工具调用安全Agent调用的工具如写文件、执行命令、访问数据库必须有严格的权限控制和沙箱机制。数据隐私上传至云端API的提示词和私有知识片段需关注服务商的隐私政策。对敏感数据应考虑本地模型部署。输出审核任何面向用户的生产环境都必须对AI生成的内容进行安全与合规性审核。3. 环境准备与前置条件在运行示例代码前你需要准备好以下环境。我们的演示将基于OpenAI API和LangChain/LangGraph生态。基础环境清单操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 20.04 推荐)。本文示例在 Linux/macOS 终端或 Windows WSL2 下测试。Python版本 3.10 或 3.11。避免使用 3.12 可能存在的兼容性问题。包管理工具pip最新版。网络能够访问https://api.openai.com(或你使用的其他大模型供应商API)。代码编辑器VS Code, PyCharm 等。核心Python包我们将使用langchain、langgraph、langchain-openai等核心库以及用于RAG的向量数据库chromadb。API密钥你需要一个有效的OpenAI API Key。也可以在代码中替换为其他兼容OpenAI API的模型服务如 DeepSeek, Together AI, 本地部署的vLLM服务等。4. 安装部署与启动方式首先创建一个新的项目目录并安装必要的依赖。# 1. 创建项目目录并进入 mkdir ai-agent-demo cd ai-agent-demo # 2. 创建并激活虚拟环境 (推荐) python -m venv venv # Windows: venv\Scripts\activate # Linux/macOS: source venv/bin/activate # 3. 安装核心依赖 pip install langchain langgraph langchain-openai langchain-chroma tiktoken # langchain-chroma 是LangChain对ChromaDB的集成包 # tiktoken 用于OpenAI模型的token计数接下来创建一个名为main.py的Python文件我们将在此构建一个整合了RAG和工具调用能力的智能体。5. 功能测试与效果验证我们将构建一个“研究助手”智能体它具备两种核心能力RAG能力回答关于我们提供的私有文档例如一篇关于“LangGraph”的Markdown简介的问题。Agent能力对于需要最新信息或计算的问题能自主调用网络搜索或计算器工具。5.1 项目结构与代码实现以下是main.py的完整代码包含了详细的注释。# main.py import os from typing import List, TypedDict, Annotated import operator from langchain_openai import ChatOpenAI from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings from langchain.tools import Tool from langchain_community.tools import DuckDuckGoSearchRun from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langgraph.checkpoint.memory import MemorySaver from langchain_core.messages import HumanMessage, AIMessage # --- 0. 设置环境变量 (你的OpenAI API Key) --- os.environ[OPENAI_API_KEY] sk-你的真实API密钥 # 请务必替换 # --- 1. 定义智能体的状态结构 --- class AgentState(TypedDict): 定义Graph运行过程中的状态流。 # 消息历史 messages: Annotated[List, add_messages] # 用户当前输入的问题 user_input: str # Agent执行过程中的中间步骤可选用于调试 intermediate_steps: List # --- 2. 准备RAG知识库 --- def create_rag_vector_store(): 创建或加载一个简单的向量知识库。 # 假设我们有一个关于LangGraph的文档 knowledge_text LangGraph 是 LangChain 生态系统中的一个库用于构建有状态、多参与者的应用程序。 它基于图Graph的概念其中节点代表执行单元如LLM调用、工具调用边代表控制流。 核心优势包括支持循环和分支、内置持久化检查点Checkpoint、便于构建复杂Agent工作流。 常用类包括StateGraph, END, START。 它常用于构建具备长期记忆和工具调用能力的AI智能体Agent。 # 将文本保存为临时文件供Loader读取实际项目可直接从数据库或文件读取 with open(knowledge.txt, w, encodingutf-8) as f: f.write(knowledge_text) # 加载并分割文档 loader TextLoader(knowledge.txt, encodingutf-8) documents loader.load() text_splitter RecursiveCharacterTextSplitter(chunk_size500, chunk_overlap50) splits text_splitter.split_documents(documents) # 创建向量存储使用ChromaDB持久化到本地./chroma_db目录 vectorstore Chroma.from_documents( documentssplits, embeddingOpenAIEmbeddings(modeltext-embedding-3-small), persist_directory./chroma_db ) return vectorstore # 初始化RAG检索器 vectorstore create_rag_vector_store() retriever vectorstore.as_retriever(search_kwargs{k: 3}) # 检索最相关的3个片段 def rag_qa(query: str) - str: RAG问答函数检索相关知识后让LLM生成答案。 # 1. 检索相关文档片段 docs retriever.invoke(query) context \n\n.join([doc.page_content for doc in docs]) # 2. 构建提示词 prompt f你是一个专业的AI助手请根据以下上下文信息回答问题。 如果上下文信息不足以回答问题请直接说“根据已有信息无法回答此问题”不要编造信息。 上下文信息 {context} 用户问题{query} 请给出准确、简洁的回答 # 3. 调用LLM llm ChatOpenAI(modelgpt-4o-mini, temperature0) response llm.invoke(prompt) return response.content # --- 3. 定义工具集 --- # 工具1: RAG工具 rag_tool Tool( nameKnowledge_Base_QA, funcrag_qa, description当用户询问关于LangGraph、RAG、Agent等本项目相关技术概念时使用此工具从知识库中寻找答案。 ) # 工具2: 网络搜索工具 search_tool DuckDuckGoSearchRun(nameWeb_Search) # 工具3: 计算器工具示例 def calculator(expression: str) - str: 一个简单的计算器工具。警告使用eval有安全风险仅用于演示。生产环境需替换。 try: result eval(expression) return str(result) except Exception as e: return f计算错误{e} calc_tool Tool( nameCalculator, funccalculator, description用于执行数学计算。输入一个有效的数学表达式如 (12 5) * 3。 ) tools [rag_tool, search_tool, calc_tool] # --- 4. 创建标准LangChain Agent --- llm ChatOpenAI(modelgpt-4o-mini, temperature0) prompt ChatPromptTemplate.from_messages([ (system, 你是一个强大的研究助手。你可以使用工具来回答问题。请根据情况选择最合适的工具。 对于明确的技术概念问题优先使用知识库工具。对于需要最新信息的问题使用网络搜索。 回答要清晰、准确。), MessagesPlaceholder(variable_namechat_history), (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ]) agent create_tool_calling_agent(llm, tools, prompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue) # --- 5. 构建LangGraph工作流 --- memory MemorySaver() # 内存检查点用于保存对话状态 def call_agent(state: AgentState): 节点函数调用Agent执行工具并生成回答。 # 准备输入 input_dict {input: state[user_input], chat_history: state[messages]} # 执行Agent response agent_executor.invoke(input_dict) # 更新消息历史 new_messages state[messages] [HumanMessage(contentstate[user_input]), AIMessage(contentresponse[output])] return {messages: new_messages, intermediate_steps: []} # 创建图 workflow StateGraph(AgentState) # 添加节点这里我们简化成一个核心Agent节点 workflow.add_node(agent, call_agent) # 设置入口点 workflow.set_entry_point(agent) # 设置结束点 workflow.add_edge(agent, END) # 编译图并启用记忆检查点 app workflow.compile(checkpointermemory) # --- 6. 测试函数 --- def test_agent(): 测试智能体的不同能力。 config {configurable: {thread_id: test_thread_1}} test_cases [ LangGraph是什么它有什么核心优势, # 应触发RAG工具 今天北京天气怎么样, # 应触发网络搜索工具 计算一下125的平方根是多少, # 应触发计算器工具 给我讲个笑话。 # 无需工具直接由LLM回答 ] for query in test_cases: print(f\n{*50}) print(f[用户] {query}) # 初始化或恢复状态 initial_state {user_input: query, messages: [], intermediate_steps: []} # 流式输出过程verbose已在AgentExecutor中开启 final_state app.invoke(initial_state, configconfig) print(f[助手] {final_state[messages][-1].content}) print(f{*50}) if __name__ __main__: print(开始测试整合了RAG与工具调用能力的智能体...) test_agent() print(\n测试完成。对话状态已保存至内存检查点。)5.2 运行与效果验证启动测试在终端中确保虚拟环境已激活并运行脚本。python main.py预期行为与观察对于“LangGraph是什么”程序会首先调用Knowledge_Base_QA工具从我们创建的knowledge.txt文档中检索相关信息然后由LLM生成答案。你会在终端看到类似Invoking: Knowledge_Base_QA with args...的日志随后输出基于知识库的答案。对于“今天北京天气怎么样”程序会调用Web_Search工具DuckDuckGo获取实时信息并生成回答。注意网络搜索可能因网络或API限制稍慢。对于计算问题程序会调用Calculator工具进行计算。对于闲聊问题LLM将直接生成回复不调用任何工具。成功标准脚本能成功运行不报错除可能的网络搜索超时警告。针对不同类型的问题能正确选择并调用对应的工具。RAG回答应基于提供的知识文本而不是通用知识。对话状态 (thread_id: “test_thread_1”) 在多次调用app.invoke时会被保留虽然本例中每次是新的初始状态但框架支持多轮。6. 接口API与批量任务上述脚本是直接运行的。在实际应用中我们通常会将这个智能体工作流封装成API服务以便集成到Web或移动应用中。6.1 使用FastAPI封装为Web服务创建一个新的文件api.py# api.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List import uvicorn # 导入之前写好的app (需要从main.py中导入) # 假设我们将之前的代码重构将app的创建放在一个模块中这里直接导入 # 为了示例我们在这里简单重构。实际项目应分模块。 from main import app, memory # 假设main.py中已将app和memory定义为全局变量 app_fastapi FastAPI(titleAI智能体API服务) class ChatRequest(BaseModel): thread_id: str “default_thread” # 会话线程ID用于区分不同对话 message: str # 用户输入 reset: bool False # 是否重置该线程的对话历史 class ChatResponse(BaseModel): thread_id: str response: str history: List[dict] # 简化的历史记录 app_fastapi.post(“/chat”, response_modelChatResponse) async def chat_with_agent(req: ChatRequest): try: config {“configurable”: {“thread_id”: req.thread_id}} # 如果需要重置则清除该线程的检查点记忆 if req.reset: # 注意MemorySaver的clear方法可能不存在这里演示逻辑。 # 实际可使用 memory.checkpointer 相关接口或直接管理状态。 pass # 简化处理实际项目需实现状态清除逻辑 # 准备输入状态。这里需要从检查点恢复历史简化起见我们每次只发最新消息。 # LangGraph的invoke会自动从checkpoint恢复历史。 initial_state {“user_input”: req.message, “messages”: [], “intermediate_steps”: []} # 调用图工作流 result app.invoke(initial_state, configconfig) # 提取最新回复 last_message result[“messages”][-1] response_text last_message.content if hasattr(last_message, ‘content’) else str(last_message) # 格式化历史可选 history [{“role”: msg.type, “content”: msg.content} for msg in result[“messages”]] return ChatResponse( thread_idreq.thread_id, responseresponse_text, historyhistory ) except Exception as e: raise HTTPException(status_code500, detailf“智能体处理失败: {str(e)}”) app_fastapi.get(“/health”) async def health_check(): return {“status”: “ok”} if __name__ “__main__”: uvicorn.run(app_fastapi, host“0.0.0.0”, port8000)运行API服务uvicorn api:app_fastapi --reload --host 0.0.0.0 --port 8000访问http://127.0.0.1:8000/docs即可看到自动生成的Swagger文档并测试/chat接口。6.2 批量任务处理对于批量处理任务例如处理一个包含100个问题的CSV文件可以设计一个异步队列。# batch_processor.py (概念示例) import asyncio import pandas as pd from main import app # 导入核心智能体应用 async def process_batch_questions(input_csv: str, output_csv: str): 批量处理问题并将结果写入CSV。 df pd.read_csv(input_csv) results [] for index, row in df.iterrows(): question row[“question”] thread_id f“batch_{index}” # 每个问题独立线程或共用线程 config {“configurable”: {“thread_id”: thread_id}} print(f“处理中: {question}”) try: initial_state {“user_input”: question, “messages”: [], “intermediate_steps”: []} result app.invoke(initial_state, configconfig) answer result[“messages”][-1].content results.append({“question”: question, “answer”: answer}) except Exception as e: results.append({“question”: question, “answer”: f“ERROR: {str(e)}”}) # 可添加延时避免对API造成压力 await asyncio.sleep(1) # 保存结果 result_df pd.DataFrame(results) result_df.to_csv(output_csv, indexFalse) print(f“批量处理完成结果已保存至 {output_csv}”) # 使用示例 # asyncio.run(process_batch_questions(“questions.csv”, “answers.csv”))7. 资源占用与性能观察本示例的资源消耗主要分为三部分大模型API调用这是主要成本。gpt-4o-mini成本较低但频繁调用仍需关注费用和速率限制。可以在LangChain中设置max_concurrency和rate limit。本地向量数据库ChromaDB运行在本地内存占用与文档库大小成正比。一个百万级token的文档库内存占用通常在几百MB到1GB。可以通过persist_directory将索引存储在磁盘。应用运行内存Python脚本本身内存占用很小几十MB。LangGraph的状态检查点如果使用MemorySaver且对话历史很长会占用更多内存。生产环境应使用Redis或PostgreSQL等外部存储作为检查点。性能优化建议RAG检索优化调整search_kwargs{“k”: 3}中的k值平衡召回率与上下文长度。使用更好的嵌入模型如text-embedding-3-large提升检索质量。工具调用降级为网络搜索等可能超时的工具设置超时和重试机制。缓存对常见的、结果不变的RAG查询如概念定义引入缓存如langchain.cache。异步处理对于批量任务使用asyncio或Celery进行异步处理避免阻塞。8. 常见问题与排查方法问题现象可能原因排查方式解决方案导入错误No module named ‘langchain_xxx’依赖包未正确安装或版本冲突。检查pip list确认包已安装。使用虚拟环境并严格按本文的pip install命令安装。或尝试pip install -r requirements.txt。运行时报错OpenAI API相关错误API密钥未设置、无效、或余额不足。网络不通。1. 检查os.environ[“OPENAI_API_KEY”]是否设置正确。2. 在终端用curl或ping测试API连通性。1. 确保密钥正确且有效。2. 检查网络代理设置。3. 可尝试更换为其他兼容API需修改ChatOpenAI的base_url和api_key。RAG检索不到相关内容回答“无法回答”1. 知识库文档未正确加载或分割。2. 检索参数k太小或相似度阈值不合适。3. 嵌入模型不匹配。1. 检查knowledge.txt文件是否存在且内容正确。2. 打印retriever.invoke(query)的结果看返回的文档片段是否相关。1. 确保文档加载和分割过程无误。2. 调整search_kwargs如增加k或使用score_threshold。3. 确保嵌入模型与创建向量库时使用的模型一致。Agent总是选择错误的工具或直接回答而不调用工具1. 工具描述 (description) 不够清晰。2. LLM的system prompt引导性不强。3. 模型能力不足。观察verboseTrue时的日志看Agent决定调用哪个工具时的思考过程。1. 优化工具描述明确其适用场景。2. 强化system prompt中的指令例如“你必须使用工具来回答问题”。3. 尝试使用更强大的模型如gpt-4o。网络搜索工具超时或返回空网络问题或DuckDuckGo API不稳定。单独测试DuckDuckGoSearchRun().run(“test”)是否正常。1. 增加超时设置。2. 考虑替换为其他搜索工具如SerpAPI、Google Search API。3. 在工具调用外层添加try-catch提供降级回答。LangGraph状态记忆没有保存检查点配置不正确或每次调用使用了不同的thread_id。检查config字典中的thread_id是否保持一致。确保同一会话使用相同的thread_id。MemorySaver是内存存储重启服务后状态会丢失生产环境需换用持久化存储。批量处理时API达到速率限制请求频率过高。查看API返回的错误信息。在批量任务中增加延时 (await asyncio.sleep(1))或使用LangChain的RateLimiter或申请提升API速率限制。9. 最佳实践与使用建议提示词工程是基石system prompt和工具描述 (description) 的质量直接决定Agent的表现。务必清晰、具体、无歧义。可以建立提示词模板库进行管理。从简单开始逐步复杂化先验证单个工具如RAG工作正常再逐步加入更多工具和复杂的LangGraph工作流如循环、分支、人工审核节点。为生产环境做好准备检查点持久化将MemorySaver替换为Redis或PostgreSQL等后端以支持服务重启和水平扩展。可观测性集成LangSmith它可以可视化跟踪每一次LLM调用、工具调用和Graph的执行路径便于调试和优化。错误处理与降级对所有工具调用和LLM调用进行健壮的异常捕获并提供友好的用户反馈或降级方案。安全第一工具权限像Calculator中使用eval是极度危险的仅用于演示。生产环境必须使用安全的表达式解析库如ast.literal_eval或自定义安全函数。用户输入净化对传入Agent的用户输入进行必要的清洗和过滤防止提示词注入攻击。内容审核在Agent最终输出前增加一个内容安全审核节点可调用审核API或使用规则引擎。成本与性能监控密切关注Token消耗和API调用次数设置预算警报。对于RAG可以考虑对检索结果进行压缩或摘要以减少送入LLM的上下文长度从而降低成本。10. 总结与下一步这套结合了Prompt Engineering、LangGraph、RAG 和 Agent的技术栈代表了当前构建复杂AI应用的主流方向。它的价值不在于单个组件的炫技而在于提供了一套标准化、可编排、可扩展的框架让开发者能够像搭积木一样构建功能强大的AI智能体。最值得尝试的点LangGraph的工作流可视化其真正的威力在于用图来定义复杂逻辑你可以通过LangChain Studio或代码清晰地看到StateGraph的流转这比传统的线性脚本更易维护和调试。RAG与Agent的融合让Agent既拥有私有知识又能调用实时工具极大地扩展了应用边界。最先应该验证的功能 按照本文的示例你应该首先确保RAG部分能正确地从你的私有文档中检索并回答问题。Agent能根据问题类型正确选择工具RAG、搜索、计算。最容易踩的坑工具描述不清导致LLM无法正确选择工具。状态管理混乱thread_id使用不当导致对话记忆错乱。忽略错误处理网络工具失败导致整个流程崩溃。后续扩展方向集成更多工具如数据库查询、发送邮件、调用内部API等。实现多智能体协作使用LangGraph定义多个具有不同角色的Agent让他们协作完成更复杂的任务。前端界面使用Gradio或Streamlit快速构建一个Web界面与你的智能体交互。本地模型部署将OpenAI API替换为本地部署的Ollama(如Llama 3.2, Qwen2.5) 或vLLM服务实现完全私有化。建议将本文的示例代码作为起点复制到你的本地环境运行一遍理解数据流和控制流。然后尝试用你自己的文档替换knowledge.txt或者添加一个新的工具。只有亲手调试和迭代你才能真正掌握这套强大工具链的精髓。
这次我们来看一个关于提示词工程Prompt Engineering和现代大模型应用开发栈的综合性技术主题。标题中提到的“吴恩达”和“2026公认最好的”更多是吸引眼球的表述但其核心指向的是当前大模型应用开发中几个最关键的架构范式Prompt Engineering提示词工程、LangGraph、RAG检索增强生成和Agent智能体。这并非一个单一的“项目”而是一套构建复杂、可靠AI应用的方法论和工具链。对于开发者而言最关心的不是哪个概念最火而是这套技术栈能不能用、怎么用、门槛高不高、以及如何整合。本文将抛开营销术语直接切入技术核心带你快速理解这四大支柱并通过一个整合了LangGraph、RAG和Agent的实战代码示例演示如何构建一个具备记忆、工具调用和知识检索能力的智能体应用。我们将重点关注其架构设计、代码实现、运行方式以及在实际场景中的效果验证。1. 核心能力速览下表概括了本次探讨的技术栈核心组件及其在应用开发中的角色能力项说明技术栈构成Prompt Engineering LangGraph RAG Agent核心定位构建复杂、可控制、具备长期记忆和外部知识的大模型应用框架主要功能结构化提示设计、有状态的工作流编排、外部知识检索、自主工具调用与任务分解硬件门槛开发阶段普通CPU/GPU均可依赖云端大模型API如OpenAI、DeepSeek。本地部署如需本地运行大模型则需相应GPU资源。启动方式基于Python脚本启动可通过FastAPI等封装为Web API服务。是否支持API是。核心逻辑可轻松封装为RESTful或GraphQL API供前端或其他服务调用。是否支持批量/异步任务是。LangGraph的工作流和Agent任务可以设计为处理队列任务。适合场景智能客服、复杂问答系统、自动化研究助手、个性化内容生成、数据分析Agent等需要多步骤推理和知识结合的场景。简单来说这套组合拳解决了单纯调用大模型API的几大痛点提示词效果不稳定、无法执行多轮复杂任务、缺乏私有知识、以及无法操作外部工具。2. 适用场景与使用边界适合谁全栈/后端开发者希望将大模型能力深度集成到现有产品中。AI应用创业者需要快速构建具备复杂逻辑的AI原型或产品。技术团队寻求标准化、可维护的大模型应用开发框架。能解决什么问题可控的对话与任务流通过LangGraph将多轮对话、工具调用、条件分支编排成可视化的工作流避免代码 spaghetti。知识库问答通过RAG让大模型能够回答关于特定领域、私有文档如产品手册、内部Wiki的问题避免“幻觉”。自动化智能体通过Agent框架让AI能够理解用户目标自主规划步骤、调用搜索引擎、计算器、数据库等工具完成任务。提示词标准化通过Prompt Engineering最佳实践将有效的提示词模板化、参数化提升生成效果的一致性和可复用性。不适合什么场景极其简单的单次问答如果只是调用ChatCompletion一次就能解决引入全套框架反而过度设计。对延迟极其敏感的场景RAG检索、多步Agent推理会增加响应时间。完全离线的环境如果无法连接任何大模型API包括本地部署的模型则无法运行。安全与合规边界知识版权RAG使用的文档需确保拥有合法使用权避免侵犯版权。工具调用安全Agent调用的工具如写文件、执行命令、访问数据库必须有严格的权限控制和沙箱机制。数据隐私上传至云端API的提示词和私有知识片段需关注服务商的隐私政策。对敏感数据应考虑本地模型部署。输出审核任何面向用户的生产环境都必须对AI生成的内容进行安全与合规性审核。3. 环境准备与前置条件在运行示例代码前你需要准备好以下环境。我们的演示将基于OpenAI API和LangChain/LangGraph生态。基础环境清单操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 20.04 推荐)。本文示例在 Linux/macOS 终端或 Windows WSL2 下测试。Python版本 3.10 或 3.11。避免使用 3.12 可能存在的兼容性问题。包管理工具pip最新版。网络能够访问https://api.openai.com(或你使用的其他大模型供应商API)。代码编辑器VS Code, PyCharm 等。核心Python包我们将使用langchain、langgraph、langchain-openai等核心库以及用于RAG的向量数据库chromadb。API密钥你需要一个有效的OpenAI API Key。也可以在代码中替换为其他兼容OpenAI API的模型服务如 DeepSeek, Together AI, 本地部署的vLLM服务等。4. 安装部署与启动方式首先创建一个新的项目目录并安装必要的依赖。# 1. 创建项目目录并进入 mkdir ai-agent-demo cd ai-agent-demo # 2. 创建并激活虚拟环境 (推荐) python -m venv venv # Windows: venv\Scripts\activate # Linux/macOS: source venv/bin/activate # 3. 安装核心依赖 pip install langchain langgraph langchain-openai langchain-chroma tiktoken # langchain-chroma 是LangChain对ChromaDB的集成包 # tiktoken 用于OpenAI模型的token计数接下来创建一个名为main.py的Python文件我们将在此构建一个整合了RAG和工具调用能力的智能体。5. 功能测试与效果验证我们将构建一个“研究助手”智能体它具备两种核心能力RAG能力回答关于我们提供的私有文档例如一篇关于“LangGraph”的Markdown简介的问题。Agent能力对于需要最新信息或计算的问题能自主调用网络搜索或计算器工具。5.1 项目结构与代码实现以下是main.py的完整代码包含了详细的注释。# main.py import os from typing import List, TypedDict, Annotated import operator from langchain_openai import ChatOpenAI from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings from langchain.tools import Tool from langchain_community.tools import DuckDuckGoSearchRun from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langgraph.checkpoint.memory import MemorySaver from langchain_core.messages import HumanMessage, AIMessage # --- 0. 设置环境变量 (你的OpenAI API Key) --- os.environ[OPENAI_API_KEY] sk-你的真实API密钥 # 请务必替换 # --- 1. 定义智能体的状态结构 --- class AgentState(TypedDict): 定义Graph运行过程中的状态流。 # 消息历史 messages: Annotated[List, add_messages] # 用户当前输入的问题 user_input: str # Agent执行过程中的中间步骤可选用于调试 intermediate_steps: List # --- 2. 准备RAG知识库 --- def create_rag_vector_store(): 创建或加载一个简单的向量知识库。 # 假设我们有一个关于LangGraph的文档 knowledge_text LangGraph 是 LangChain 生态系统中的一个库用于构建有状态、多参与者的应用程序。 它基于图Graph的概念其中节点代表执行单元如LLM调用、工具调用边代表控制流。 核心优势包括支持循环和分支、内置持久化检查点Checkpoint、便于构建复杂Agent工作流。 常用类包括StateGraph, END, START。 它常用于构建具备长期记忆和工具调用能力的AI智能体Agent。 # 将文本保存为临时文件供Loader读取实际项目可直接从数据库或文件读取 with open(knowledge.txt, w, encodingutf-8) as f: f.write(knowledge_text) # 加载并分割文档 loader TextLoader(knowledge.txt, encodingutf-8) documents loader.load() text_splitter RecursiveCharacterTextSplitter(chunk_size500, chunk_overlap50) splits text_splitter.split_documents(documents) # 创建向量存储使用ChromaDB持久化到本地./chroma_db目录 vectorstore Chroma.from_documents( documentssplits, embeddingOpenAIEmbeddings(modeltext-embedding-3-small), persist_directory./chroma_db ) return vectorstore # 初始化RAG检索器 vectorstore create_rag_vector_store() retriever vectorstore.as_retriever(search_kwargs{k: 3}) # 检索最相关的3个片段 def rag_qa(query: str) - str: RAG问答函数检索相关知识后让LLM生成答案。 # 1. 检索相关文档片段 docs retriever.invoke(query) context \n\n.join([doc.page_content for doc in docs]) # 2. 构建提示词 prompt f你是一个专业的AI助手请根据以下上下文信息回答问题。 如果上下文信息不足以回答问题请直接说“根据已有信息无法回答此问题”不要编造信息。 上下文信息 {context} 用户问题{query} 请给出准确、简洁的回答 # 3. 调用LLM llm ChatOpenAI(modelgpt-4o-mini, temperature0) response llm.invoke(prompt) return response.content # --- 3. 定义工具集 --- # 工具1: RAG工具 rag_tool Tool( nameKnowledge_Base_QA, funcrag_qa, description当用户询问关于LangGraph、RAG、Agent等本项目相关技术概念时使用此工具从知识库中寻找答案。 ) # 工具2: 网络搜索工具 search_tool DuckDuckGoSearchRun(nameWeb_Search) # 工具3: 计算器工具示例 def calculator(expression: str) - str: 一个简单的计算器工具。警告使用eval有安全风险仅用于演示。生产环境需替换。 try: result eval(expression) return str(result) except Exception as e: return f计算错误{e} calc_tool Tool( nameCalculator, funccalculator, description用于执行数学计算。输入一个有效的数学表达式如 (12 5) * 3。 ) tools [rag_tool, search_tool, calc_tool] # --- 4. 创建标准LangChain Agent --- llm ChatOpenAI(modelgpt-4o-mini, temperature0) prompt ChatPromptTemplate.from_messages([ (system, 你是一个强大的研究助手。你可以使用工具来回答问题。请根据情况选择最合适的工具。 对于明确的技术概念问题优先使用知识库工具。对于需要最新信息的问题使用网络搜索。 回答要清晰、准确。), MessagesPlaceholder(variable_namechat_history), (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ]) agent create_tool_calling_agent(llm, tools, prompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue) # --- 5. 构建LangGraph工作流 --- memory MemorySaver() # 内存检查点用于保存对话状态 def call_agent(state: AgentState): 节点函数调用Agent执行工具并生成回答。 # 准备输入 input_dict {input: state[user_input], chat_history: state[messages]} # 执行Agent response agent_executor.invoke(input_dict) # 更新消息历史 new_messages state[messages] [HumanMessage(contentstate[user_input]), AIMessage(contentresponse[output])] return {messages: new_messages, intermediate_steps: []} # 创建图 workflow StateGraph(AgentState) # 添加节点这里我们简化成一个核心Agent节点 workflow.add_node(agent, call_agent) # 设置入口点 workflow.set_entry_point(agent) # 设置结束点 workflow.add_edge(agent, END) # 编译图并启用记忆检查点 app workflow.compile(checkpointermemory) # --- 6. 测试函数 --- def test_agent(): 测试智能体的不同能力。 config {configurable: {thread_id: test_thread_1}} test_cases [ LangGraph是什么它有什么核心优势, # 应触发RAG工具 今天北京天气怎么样, # 应触发网络搜索工具 计算一下125的平方根是多少, # 应触发计算器工具 给我讲个笑话。 # 无需工具直接由LLM回答 ] for query in test_cases: print(f\n{*50}) print(f[用户] {query}) # 初始化或恢复状态 initial_state {user_input: query, messages: [], intermediate_steps: []} # 流式输出过程verbose已在AgentExecutor中开启 final_state app.invoke(initial_state, configconfig) print(f[助手] {final_state[messages][-1].content}) print(f{*50}) if __name__ __main__: print(开始测试整合了RAG与工具调用能力的智能体...) test_agent() print(\n测试完成。对话状态已保存至内存检查点。)5.2 运行与效果验证启动测试在终端中确保虚拟环境已激活并运行脚本。python main.py预期行为与观察对于“LangGraph是什么”程序会首先调用Knowledge_Base_QA工具从我们创建的knowledge.txt文档中检索相关信息然后由LLM生成答案。你会在终端看到类似Invoking: Knowledge_Base_QA with args...的日志随后输出基于知识库的答案。对于“今天北京天气怎么样”程序会调用Web_Search工具DuckDuckGo获取实时信息并生成回答。注意网络搜索可能因网络或API限制稍慢。对于计算问题程序会调用Calculator工具进行计算。对于闲聊问题LLM将直接生成回复不调用任何工具。成功标准脚本能成功运行不报错除可能的网络搜索超时警告。针对不同类型的问题能正确选择并调用对应的工具。RAG回答应基于提供的知识文本而不是通用知识。对话状态 (thread_id: “test_thread_1”) 在多次调用app.invoke时会被保留虽然本例中每次是新的初始状态但框架支持多轮。6. 接口API与批量任务上述脚本是直接运行的。在实际应用中我们通常会将这个智能体工作流封装成API服务以便集成到Web或移动应用中。6.1 使用FastAPI封装为Web服务创建一个新的文件api.py# api.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List import uvicorn # 导入之前写好的app (需要从main.py中导入) # 假设我们将之前的代码重构将app的创建放在一个模块中这里直接导入 # 为了示例我们在这里简单重构。实际项目应分模块。 from main import app, memory # 假设main.py中已将app和memory定义为全局变量 app_fastapi FastAPI(titleAI智能体API服务) class ChatRequest(BaseModel): thread_id: str “default_thread” # 会话线程ID用于区分不同对话 message: str # 用户输入 reset: bool False # 是否重置该线程的对话历史 class ChatResponse(BaseModel): thread_id: str response: str history: List[dict] # 简化的历史记录 app_fastapi.post(“/chat”, response_modelChatResponse) async def chat_with_agent(req: ChatRequest): try: config {“configurable”: {“thread_id”: req.thread_id}} # 如果需要重置则清除该线程的检查点记忆 if req.reset: # 注意MemorySaver的clear方法可能不存在这里演示逻辑。 # 实际可使用 memory.checkpointer 相关接口或直接管理状态。 pass # 简化处理实际项目需实现状态清除逻辑 # 准备输入状态。这里需要从检查点恢复历史简化起见我们每次只发最新消息。 # LangGraph的invoke会自动从checkpoint恢复历史。 initial_state {“user_input”: req.message, “messages”: [], “intermediate_steps”: []} # 调用图工作流 result app.invoke(initial_state, configconfig) # 提取最新回复 last_message result[“messages”][-1] response_text last_message.content if hasattr(last_message, ‘content’) else str(last_message) # 格式化历史可选 history [{“role”: msg.type, “content”: msg.content} for msg in result[“messages”]] return ChatResponse( thread_idreq.thread_id, responseresponse_text, historyhistory ) except Exception as e: raise HTTPException(status_code500, detailf“智能体处理失败: {str(e)}”) app_fastapi.get(“/health”) async def health_check(): return {“status”: “ok”} if __name__ “__main__”: uvicorn.run(app_fastapi, host“0.0.0.0”, port8000)运行API服务uvicorn api:app_fastapi --reload --host 0.0.0.0 --port 8000访问http://127.0.0.1:8000/docs即可看到自动生成的Swagger文档并测试/chat接口。6.2 批量任务处理对于批量处理任务例如处理一个包含100个问题的CSV文件可以设计一个异步队列。# batch_processor.py (概念示例) import asyncio import pandas as pd from main import app # 导入核心智能体应用 async def process_batch_questions(input_csv: str, output_csv: str): 批量处理问题并将结果写入CSV。 df pd.read_csv(input_csv) results [] for index, row in df.iterrows(): question row[“question”] thread_id f“batch_{index}” # 每个问题独立线程或共用线程 config {“configurable”: {“thread_id”: thread_id}} print(f“处理中: {question}”) try: initial_state {“user_input”: question, “messages”: [], “intermediate_steps”: []} result app.invoke(initial_state, configconfig) answer result[“messages”][-1].content results.append({“question”: question, “answer”: answer}) except Exception as e: results.append({“question”: question, “answer”: f“ERROR: {str(e)}”}) # 可添加延时避免对API造成压力 await asyncio.sleep(1) # 保存结果 result_df pd.DataFrame(results) result_df.to_csv(output_csv, indexFalse) print(f“批量处理完成结果已保存至 {output_csv}”) # 使用示例 # asyncio.run(process_batch_questions(“questions.csv”, “answers.csv”))7. 资源占用与性能观察本示例的资源消耗主要分为三部分大模型API调用这是主要成本。gpt-4o-mini成本较低但频繁调用仍需关注费用和速率限制。可以在LangChain中设置max_concurrency和rate limit。本地向量数据库ChromaDB运行在本地内存占用与文档库大小成正比。一个百万级token的文档库内存占用通常在几百MB到1GB。可以通过persist_directory将索引存储在磁盘。应用运行内存Python脚本本身内存占用很小几十MB。LangGraph的状态检查点如果使用MemorySaver且对话历史很长会占用更多内存。生产环境应使用Redis或PostgreSQL等外部存储作为检查点。性能优化建议RAG检索优化调整search_kwargs{“k”: 3}中的k值平衡召回率与上下文长度。使用更好的嵌入模型如text-embedding-3-large提升检索质量。工具调用降级为网络搜索等可能超时的工具设置超时和重试机制。缓存对常见的、结果不变的RAG查询如概念定义引入缓存如langchain.cache。异步处理对于批量任务使用asyncio或Celery进行异步处理避免阻塞。8. 常见问题与排查方法问题现象可能原因排查方式解决方案导入错误No module named ‘langchain_xxx’依赖包未正确安装或版本冲突。检查pip list确认包已安装。使用虚拟环境并严格按本文的pip install命令安装。或尝试pip install -r requirements.txt。运行时报错OpenAI API相关错误API密钥未设置、无效、或余额不足。网络不通。1. 检查os.environ[“OPENAI_API_KEY”]是否设置正确。2. 在终端用curl或ping测试API连通性。1. 确保密钥正确且有效。2. 检查网络代理设置。3. 可尝试更换为其他兼容API需修改ChatOpenAI的base_url和api_key。RAG检索不到相关内容回答“无法回答”1. 知识库文档未正确加载或分割。2. 检索参数k太小或相似度阈值不合适。3. 嵌入模型不匹配。1. 检查knowledge.txt文件是否存在且内容正确。2. 打印retriever.invoke(query)的结果看返回的文档片段是否相关。1. 确保文档加载和分割过程无误。2. 调整search_kwargs如增加k或使用score_threshold。3. 确保嵌入模型与创建向量库时使用的模型一致。Agent总是选择错误的工具或直接回答而不调用工具1. 工具描述 (description) 不够清晰。2. LLM的system prompt引导性不强。3. 模型能力不足。观察verboseTrue时的日志看Agent决定调用哪个工具时的思考过程。1. 优化工具描述明确其适用场景。2. 强化system prompt中的指令例如“你必须使用工具来回答问题”。3. 尝试使用更强大的模型如gpt-4o。网络搜索工具超时或返回空网络问题或DuckDuckGo API不稳定。单独测试DuckDuckGoSearchRun().run(“test”)是否正常。1. 增加超时设置。2. 考虑替换为其他搜索工具如SerpAPI、Google Search API。3. 在工具调用外层添加try-catch提供降级回答。LangGraph状态记忆没有保存检查点配置不正确或每次调用使用了不同的thread_id。检查config字典中的thread_id是否保持一致。确保同一会话使用相同的thread_id。MemorySaver是内存存储重启服务后状态会丢失生产环境需换用持久化存储。批量处理时API达到速率限制请求频率过高。查看API返回的错误信息。在批量任务中增加延时 (await asyncio.sleep(1))或使用LangChain的RateLimiter或申请提升API速率限制。9. 最佳实践与使用建议提示词工程是基石system prompt和工具描述 (description) 的质量直接决定Agent的表现。务必清晰、具体、无歧义。可以建立提示词模板库进行管理。从简单开始逐步复杂化先验证单个工具如RAG工作正常再逐步加入更多工具和复杂的LangGraph工作流如循环、分支、人工审核节点。为生产环境做好准备检查点持久化将MemorySaver替换为Redis或PostgreSQL等后端以支持服务重启和水平扩展。可观测性集成LangSmith它可以可视化跟踪每一次LLM调用、工具调用和Graph的执行路径便于调试和优化。错误处理与降级对所有工具调用和LLM调用进行健壮的异常捕获并提供友好的用户反馈或降级方案。安全第一工具权限像Calculator中使用eval是极度危险的仅用于演示。生产环境必须使用安全的表达式解析库如ast.literal_eval或自定义安全函数。用户输入净化对传入Agent的用户输入进行必要的清洗和过滤防止提示词注入攻击。内容审核在Agent最终输出前增加一个内容安全审核节点可调用审核API或使用规则引擎。成本与性能监控密切关注Token消耗和API调用次数设置预算警报。对于RAG可以考虑对检索结果进行压缩或摘要以减少送入LLM的上下文长度从而降低成本。10. 总结与下一步这套结合了Prompt Engineering、LangGraph、RAG 和 Agent的技术栈代表了当前构建复杂AI应用的主流方向。它的价值不在于单个组件的炫技而在于提供了一套标准化、可编排、可扩展的框架让开发者能够像搭积木一样构建功能强大的AI智能体。最值得尝试的点LangGraph的工作流可视化其真正的威力在于用图来定义复杂逻辑你可以通过LangChain Studio或代码清晰地看到StateGraph的流转这比传统的线性脚本更易维护和调试。RAG与Agent的融合让Agent既拥有私有知识又能调用实时工具极大地扩展了应用边界。最先应该验证的功能 按照本文的示例你应该首先确保RAG部分能正确地从你的私有文档中检索并回答问题。Agent能根据问题类型正确选择工具RAG、搜索、计算。最容易踩的坑工具描述不清导致LLM无法正确选择工具。状态管理混乱thread_id使用不当导致对话记忆错乱。忽略错误处理网络工具失败导致整个流程崩溃。后续扩展方向集成更多工具如数据库查询、发送邮件、调用内部API等。实现多智能体协作使用LangGraph定义多个具有不同角色的Agent让他们协作完成更复杂的任务。前端界面使用Gradio或Streamlit快速构建一个Web界面与你的智能体交互。本地模型部署将OpenAI API替换为本地部署的Ollama(如Llama 3.2, Qwen2.5) 或vLLM服务实现完全私有化。建议将本文的示例代码作为起点复制到你的本地环境运行一遍理解数据流和控制流。然后尝试用你自己的文档替换knowledge.txt或者添加一个新的工具。只有亲手调试和迭代你才能真正掌握这套强大工具链的精髓。
)
