在实际的大模型应用开发中Prompt Engineering提示词工程是连接人类意图与模型能力的核心桥梁。它远不止是“如何提问”而是一套系统化的方法用于设计、优化和评估引导大语言模型LLM完成特定任务的指令、上下文和约束。随着应用从简单的问答走向复杂的多步骤工作流开发者需要更强大的框架来编排LLM、工具、记忆和决策逻辑。LangGraph作为LangChain生态中用于构建有状态、多参与者工作流的新星与RAG检索增强生成和Agent智能体技术结合正在成为构建下一代生成式AI应用的事实标准。本文旨在为有一定Python和LLM基础的开发者提供一个从理论到实践的深度指南。我们将首先厘清Prompt Engineering的核心原则然后以LangGraph为框架逐步构建一个具备长期记忆、工具调用和条件分支能力的智能体并集成RAG系统来增强其知识库。最终你将获得一个可运行、可扩展的智能体原型理解其内部状态流转并掌握排查常见问题与优化性能的关键技巧。1. 理解提示词工程超越简单问答的指令设计提示词工程的目标是最大化LLM在特定任务上的性能、可靠性和可控性。它不是一个静态的模板而是一个动态的优化过程。1.1 核心要素与设计模式一个高效的提示词通常包含以下几个结构化部分角色与任务定义明确指定模型在对话中扮演的角色如“资深Python开发顾问”和需要完成的具体任务。上下文与背景信息提供完成任务所需的必要信息这可以包括用户输入、从数据库或向量库检索到的相关文档、历史对话记录等。指令与步骤清晰、无歧义地列出模型需要遵循的步骤。对于复杂任务分步指令比单一复杂指令更有效。输出格式约束明确规定模型输出的格式例如JSON、Markdown、特定结构的文本甚至直接是代码。这极大地方便了后续的程序化处理。示例Few-shot Learning提供少量输入-输出示例让模型通过类比来理解任务要求这对于格式复杂或定义模糊的任务尤其有效。一个结合了以上要素的提示词示例用于文本摘要可能如下所示你是一位专业的编辑助理擅长将长篇文章浓缩为简洁的要点。 请根据用户提供的文章生成一份摘要。 # 文章 {article_text} # 要求 1. 摘要需包含原文的核心论点与关键证据。 2. 使用 bullet points 列出不超过5点。 3. 语言保持客观、中立。 4. 总字数控制在150字以内。 # 输出格式 请严格按照以下JSON格式输出 { summary_points: [要点一, 要点二, ...], word_count: 数字 }1.2 从静态提示到动态提示与思维链在智能体或复杂工作流中提示词往往是动态生成的。例如在RAG流程中系统会根据用户问题从向量库检索出相关文档片段然后将这些片段作为上下文动态插入到提示词模板中。更高级的技巧是引导模型进行“思维链”推理。通过在其思考过程中加入“让我们一步步思考”或“首先分析问题...”等指令可以显著提升模型在数学、逻辑推理等复杂任务上的表现。在LangGraph中这种多步思考过程可以通过多个节点Node和条件边Conditional Edge来显式地建模和控制。2. 环境准备与核心工具栈在开始构建智能体之前需要搭建一个稳定的开发环境。以下是我们将使用的主要工具及其版本建议。2.1 Python环境与包管理建议使用Python 3.10或更高版本并使用venv或conda创建独立的虚拟环境。# 创建并激活虚拟环境 (以 venv 为例) python -m venv langgraph-agent-env source langgraph-agent-env/bin/activate # Linux/macOS # langgraph-agent-env\Scripts\activate # Windows # 升级包管理工具 pip install --upgrade pip setuptools wheel2.2 核心依赖安装我们将安装LangChain、LangGraph、向量数据库客户端、大模型API SDK等。# 核心框架 pip install langchain langchain-community langgraph # 用于连接OpenAI、Anthropic等模型API (以OpenAI为例) pip install openai # 用于文本嵌入和向量存储 (以Chroma为例) pip install chromadb # 用于文档加载与处理 pip install pypdf python-dotenv tiktoken # 可选用于更美观的Graph可视化 pip install pyvis2.3 配置API密钥与环境变量为了安全地管理API密钥使用.env文件。在项目根目录创建.env文件。填入你的API密钥以OpenAI为例# .env OPENAI_API_KEYsk-your-openai-api-key-here # 如需其他模型可添加如 ANTHROPIC_API_KEY, GROQ_API_KEY 等在Python代码中通过dotenv加载from dotenv import load_dotenv import os load_dotenv() openai_api_key os.getenv(OPENAI_API_KEY)2.4 工具版本兼容性说明不同版本库的API可能有变化。以下是撰写本文时测试兼容的版本组合可作为参考库名称推荐版本主要用途langchain0.1.0LangChain核心框架langgraph0.0.40构建有状态工作流openai1.0.0调用GPT系列模型chromadb0.4.22轻量级向量数据库python-dotenv1.0.0环境变量管理注意LangChain生态更新较快若遇到API不兼容错误请查阅对应版本的官方文档。生产环境中建议使用requirements.txt或pyproject.toml严格锁定依赖版本。3. 构建基础LangGraph中的状态与工作流LangGraph的核心思想是将应用建模为一个有向图其中节点Node代表执行单元如调用LLM、运行工具边Edge代表状态流转的方向。图是有状态的这意味着数据状态在图执行过程中被传递和修改。3.1 定义状态State状态是一个字典或Pydantic模型包含了工作流执行过程中所有需要传递和更新的数据。我们定义一个基础状态from typing import TypedDict, List, Annotated import operator class AgentState(TypedDict): 智能体工作流的状态定义 # 用户输入的问题 question: str # 从向量库检索到的相关文档 retrieved_docs: List[str] # LLM生成的回答 answer: str # 记录LLM的思考过程或中间步骤 reasoning: List[str] # 记录已调用过的工具及其结果 tool_calls: List[dict] # 控制流程的标记如决定下一步是“回答”还是“继续检索” next_step: strTypedDict提供了类型提示。Annotated可用于更复杂的操作例如使用operator.add来合并列表。3.2 创建节点Node与边Edge节点是普通的Python函数它接收当前状态执行操作如调用LLM、查询数据库并返回更新后的状态片段。让我们创建一个简单的“检索”节点和一个“生成”节点。from langchain_openai import ChatOpenAI from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings from langchain.schema import Document # 初始化LLM和嵌入模型 llm ChatOpenAI(modelgpt-4-turbo-preview, temperature0) embeddings OpenAIEmbeddings(modeltext-embedding-3-small) # 假设我们已经有一个已加载文档的向量库 vectorstore Chroma(persist_directory./chroma_db, embedding_functionembeddings) retriever vectorstore.as_retriever(search_kwargs{k: 3}) def retrieve_node(state: AgentState) - dict: 检索节点根据问题从向量库获取相关文档 question state[question] # 执行检索 docs retriever.invoke(question) # 将文档内容提取为字符串列表 doc_contents [doc.page_content for doc in docs] # 返回要更新到状态中的字段 return {retrieved_docs: doc_contents, reasoning: [f检索到 {len(docs)} 篇相关文档。]} def generate_answer_node(state: AgentState) - dict: 生成节点基于问题和检索到的文档生成最终答案 question state[question] docs state[retrieved_docs] # 构建动态提示词 prompt f 你是一个知识渊博的助手请基于以下背景信息回答用户的问题。 如果背景信息不足以回答问题请如实告知并尝试给出一般性建议。 # 背景信息 {chr(10).join(docs)} # 用户问题 {question} # 要求 回答需准确、简洁并引用背景信息中的内容如果适用。 # 调用LLM response llm.invoke(prompt) answer response.content # 更新状态 return { answer: answer, reasoning: state[reasoning] [f基于 {len(docs)} 篇文档生成了答案。], next_step: end # 标记流程结束 }边决定了执行完一个节点后下一步该去哪个节点。最简单的边是顺序连接。在LangGraph中我们通过add_edge方法建立固定连接或使用add_conditional_edges建立条件分支。3.3 编译并运行图Graph我们将节点和边组装起来编译成一个可执行的工作流。from langgraph.graph import StateGraph, END # 1. 创建图构建器并指定状态结构 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(retrieve, retrieve_node) workflow.add_node(generate, generate_answer_node) # 3. 添加边建立固定流程retrieve - generate - END workflow.set_entry_point(retrieve) # 设置入口节点 workflow.add_edge(retrieve, generate) workflow.add_edge(generate, END) # END是LangGraph预定义的结束点 # 4. 编译图 app workflow.compile()现在我们可以运行这个简单的RAG工作流了。# 定义初始状态 initial_state {question: LangGraph的主要用途是什么, retrieved_docs: [], answer: , reasoning: [], tool_calls: [], next_step: } # 运行图 final_state app.invoke(initial_state) print(最终答案, final_state[answer]) print(推理过程, final_state[reasoning])这个基础流程实现了最直接的RAG检索 - 生成。然而真正的智能体需要判断、循环和工具调用能力。4. 实现智能体工具调用、条件分支与循环智能体的核心是能够根据情况自主决定下一步行动。在LangGraph中这通过条件边和工具调用节点来实现。4.1 为LLM装备工具Tools工具是智能体与外界交互的接口可以是搜索、计算、数据库查询等任何函数。from langchain.tools import tool from datetime import datetime tool def get_current_time(tz: str Asia/Shanghai) - str: 获取指定时区的当前时间。 # 这是一个简化实现实际应用中应使用pytz等库 now datetime.now() return now.strftime(f%Y-%m-%d %H:%M:%S (假设时区: {tz})) tool def web_search(query: str) - str: 模拟网络搜索。在生产环境中这里应接入真实的搜索API。 # 模拟返回 return f关于 {query} 的模拟搜索结果相关链接1相关链接2。 # 将工具列表提供给LLM tools [get_current_time, web_search] llm_with_tools llm.bind_tools(tools)4.2 创建工具调用与路由逻辑我们需要一个节点来处理LLM的决策是直接回答还是调用工具from langchain_core.messages import AIMessage, HumanMessage, ToolMessage def agent_node(state: AgentState) - dict: 智能体决策节点决定调用工具还是直接回答。 question state[question] # 将对话历史简化和当前问题组成消息列表 messages [HumanMessage(contentquestion)] # 调用绑定了工具的LLM response llm_with_tools.invoke(messages) # 初始化返回的更新字段 updates {reasoning: state[reasoning] [fLLM响应类型: {type(response).__name__}]} # 判断响应类型 if isinstance(response, AIMessage) and response.tool_calls: # LLM决定调用工具 tool_calls response.tool_calls updates[tool_calls] state[tool_calls] tool_calls updates[next_step] call_tools # 下一步去执行工具 # 将LLM的消息也记录下来方便后续构造对话历史 updates[last_ai_message] response else: # LLM决定直接回答 updates[answer] response.content updates[next_step] end # 流程结束 return updates然后我们需要一个节点来实际执行被调用的工具。def tool_node(state: AgentState) - dict: 工具执行节点运行LLM请求的工具并收集结果。 tool_calls state[tool_calls] last_message state.get(last_ai_message) tool_messages [] for tc in tool_calls[-1:]: # 通常只执行最新的一组工具调用 tool_name tc[name] tool_args tc[args] # 根据工具名找到对应的工具函数 tool_to_use next((t for t in tools if t.name tool_name), None) if tool_to_use: try: result tool_to_use.invoke(tool_args) tool_messages.append(ToolMessage(contentstr(result), tool_call_idtc[id])) except Exception as e: tool_messages.append(ToolMessage(contentfError: {e}, tool_call_idtc[id])) else: tool_messages.append(ToolMessage(contentfTool {tool_name} not found., tool_call_idtc[id])) # 更新状态记录工具执行结果并决定下一步是返回给Agent继续思考 updates { tool_results: tool_messages, next_step: agent # 执行完工具后返回智能体节点进行下一步决策 } return updates4.3 构建有条件分支的图现在我们构建一个更复杂的图它包含循环智能体可以多次决定调用工具。# 重新定义状态增加必要字段 class AgentState(TypedDict): question: str reasoning: List[str] tool_calls: List[dict] tool_results: List last_ai_message: Any answer: str next_step: str # 创建新图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(agent, agent_node) # 决策节点 workflow.add_node(tools, tool_node) # 工具执行节点 # 设置入口点 workflow.set_entry_point(agent) # 添加条件边根据 next_step 的值决定路由 from langgraph.graph import END def route_after_agent(state: AgentState) - str: 路由函数检查状态中的 next_step 字段 return state[next_step] # 返回的值必须是已定义节点的名字或 END workflow.add_conditional_edges( agent, # 源节点 route_after_agent, # 路由判断函数 { call_tools: tools, # 如果 next_step call_tools 则前往 tools 节点 end: END # 如果 next_step end 则结束 } ) # 从 tools 节点执行完后固定返回 agent 节点 workflow.add_edge(tools, agent) # 编译图 agent_app workflow.compile()4.4 运行智能体现在我们可以运行这个具备工具调用能力的智能体了。initial_state { question: 现在上海是什么时间顺便搜索一下今天的科技新闻。, reasoning: [], tool_calls: [], tool_results: [], last_ai_message: None, answer: , next_step: } # 设置最大步数以防止无限循环 from langgraph.checkpoint import MemorySaver from langgraph.graph import MessagesState # 为了支持更复杂的对话历史可以使用LangGraph预定义的MessagesState # 这里为了示例清晰我们仍使用简化状态。 # 在实际复杂应用中建议使用MessagesState或Pydantic来管理消息列表。 final_state agent_app.invoke(initial_state, config{recursion_limit: 10}) print(智能体最终答案, final_state.get(answer, 未生成最终答案)) print(工具调用记录, final_state.get(tool_calls))这个智能体会先尝试调用get_current_time工具获取时间然后可能再调用web_search工具搜索新闻最后综合所有工具结果生成最终回答。recursion_limit参数防止了因逻辑错误导致的无限循环。5. 集成RAG构建具有知识库的智能体将RAG与智能体结合意味着智能体在回答问题时可以主动从知识库中检索信息作为依据。我们可以将之前的retrieve_node整合到智能体的决策循环中。5.1 设计支持RAG的智能体流程一种常见的设计是智能体首先判断是否需要检索知识库。如果需要则进入检索节点检索完成后带着检索结果重新进入决策节点此时LLM可以基于检索到的文档来回答或决定下一步行动。我们需要修改状态和节点逻辑class RagAgentState(TypedDict): messages: Annotated[list, operator.add] # 使用LangGraph推荐的注解方式管理消息历史 question: str retrieved_docs: List[str] next_step: str def should_retrieve(state: RagAgentState) - str: 判断节点LLM判断是否需要检索知识库。 # 这里简化处理实际中可以训练一个分类器或设计更复杂的提示词让LLM判断 # 例如如果问题涉及特定内部知识则检索 prompt f 用户的问题是{state[question]} 你需要判断回答这个问题是否需要查询内部知识库 如果你认为需要请回复“retrieve”。 如果你认为不需要例如是寒暄、通用知识或工具调用请回复“reason”。 response llm.invoke(prompt) decision response.content.strip().lower() return retrieve if retrieve in decision else reason def rag_retrieve_node(state: RagAgentState) - dict: RAG检索节点 docs retriever.invoke(state[question]) doc_contents [doc.page_content for doc in docs] # 将检索结果添加到消息历史中供后续节点使用 new_message HumanMessage(contentf[检索到的背景信息]{chr(10).join(doc_contents)}) return {retrieved_docs: doc_contents, messages: [new_message]} def rag_agent_reason_node(state: RagAgentState) - dict: 推理节点基于当前所有信息可能包含检索结果进行思考或回答。 # 此节点可以集成之前的工具调用和最终回答逻辑 # 为了简化这里假设它直接生成最终答案 all_context \n.join([state[question]] state[retrieved_docs]) prompt f请根据以下信息回答问题\n{all_context} response llm.invoke(prompt) # 将回答添加到消息历史 ai_message AIMessage(contentresponse.content) return {messages: [ai_message], next_step: end}5.2 构建RAG智能体图workflow StateGraph(RagAgentState) workflow.add_node(should_retrieve, should_retrieve) # 判断节点 workflow.add_node(retrieve, rag_retrieve_node) workflow.add_node(reason, rag_agent_reason_node) workflow.set_entry_point(should_retrieve) # 从判断节点出发根据返回值路由 workflow.add_conditional_edges( should_retrieve, lambda state: state[next_step] if next_step in state else retrieve, # 简化路由逻辑 {retrieve: retrieve, reason: reason} ) workflow.add_edge(retrieve, reason) # 检索完后去推理 workflow.add_edge(reason, END) rag_agent_app workflow.compile()这个流程实现了基本的条件化RAG先判断再检索最后生成。在实际项目中reason节点可以替换为前面章节中更复杂的、具备工具调用能力的agent_node从而形成一个功能完整的“检索增强型智能体”。6. 运行验证、问题排查与性能优化构建完应用后系统的验证、监控和优化至关重要。6.1 运行验证与结果分析运行应用后不能仅看最终输出还需要检查中间状态和日志。# 使用invoke的详细模式或通过自定义回调记录 from langchain_core.callbacks import StdOutCallbackHandler final_state rag_agent_app.invoke( {question: LangGraph中如何实现循环, messages: [], retrieved_docs: [], next_step: }, config{callbacks: [StdOutCallbackHandler()]} # 打印内部事件 ) # 手动检查关键状态 print(\n 状态分析 ) print(f问题: {final_state.get(question)}) print(f检索到的文档数: {len(final_state.get(retrieved_docs, []))}) print(f最终消息历史长度: {len(final_state.get(messages, []))}) if final_state.get(messages): last_msg final_state[messages][-1] print(f最终输出: {last_msg.content if hasattr(last_msg, content) else last_msg})6.2 常见问题排查清单在开发LangGraph智能体时以下问题是高频出现的问题现象可能原因检查点与解决方案图编译失败状态结构定义与节点返回值不匹配节点函数签名错误。1. 检查StateGraph初始化时传入的状态类/字典是否与每个节点返回的字典键匹配。2. 确保所有节点函数都接收一个状态参数并返回一个字典。无限循环条件边逻辑错误导致节点间形成死循环未设置recursion_limit。1. 仔细检查add_conditional_edges的路由函数确保所有可能输出都对应有效的节点或END。2. 在app.invoke()中显式设置config{recursion_limit: N}。3. 在状态中添加iteration_count字段并在节点中递增达到阈值后强制跳转到END。工具调用不被识别LLM未正确绑定工具工具定义格式不符合LangChain要求提示词未引导LLM使用工具。1. 使用llm.bind_tools(tools)确保工具绑定成功。2. 检查tool装饰器是否正确使用工具函数是否有文档字符串docstring。3. 在系统提示词中明确告知LLM可用的工具及其用途。RAG检索结果不相关嵌入模型不匹配向量库索引未正确构建检索参数如k值不合适文本分块策略不佳。1. 确保查询时使用的嵌入模型与构建向量库时相同。2. 检查向量库中是否已成功存入文档。3. 调整retriever.search_kwargs如k,score_threshold。4. 优化文档分块chunk的大小和重叠overlap。状态更新不符合预期使用了错误的注解如Annotated在节点中直接修改了传入的状态字典应返回新字典。1. 对于列表合并等操作使用Annotated[list, operator.add]定义状态字段。2. 遵循函数式编程思想节点函数应返回一个包含更新字段的字典而不是修改原状态。图可视化混乱节点和边过多逻辑复杂。1. 使用workflow.get_graph().draw_mermaid_png()生成流程图需安装mermaid相关依赖。2. 将复杂图拆分为多个子图Subgraph进行模块化管理。6.3 性能与生产环境最佳实践异步支持LangGraph支持异步节点。对于IO密集型操作如网络请求、数据库查询使用async def定义节点函数并使用ainvoke运行图可以显著提升吞吐量。async def async_retrieve_node(state: State): # 异步检索操作 docs await retriever.ainvoke(state[question]) return {docs: docs}持久化检查点使用MemorySaver或SqliteSaver等检查点存储器可以暂停和恢复长时间运行的工作流这对于处理复杂、多轮交互的任务至关重要。from langgraph.checkpoint import MemorySaver memory MemorySaver() app workflow.compile(checkpointermemory) # 可以通过 thread_id 来管理不同的会话流 config {configurable: {thread_id: user_123}} app.invoke(initial_state, configconfig)可观测性与监控集成LangSmith可以追踪每次图的执行、每个节点的输入输出、耗时和Token使用情况便于调试和优化。import os os.environ[LANGCHAIN_TRACING_V2] true os.environ[LANGCHAIN_API_KEY] your-langsmith-api-key # 调用将被自动记录到LangSmith提示词优化与版本控制将提示词模板外置到配置文件或数据库中便于A/B测试和迭代更新。避免将长篇提示词硬编码在代码中。错误处理与降级在节点函数内部使用try...except包裹核心逻辑并更新状态以反映错误。可以设计一个专门的“错误处理”节点来接管出错的状态提供降级响应。向量库优化分块策略根据文档类型调整分块大小如技术文档256-512词小说1024词和重叠区域10-20%。元数据过滤在检索时利用元数据如文档来源、章节、日期进行过滤提高精度。重排序在初步检索召回后使用一个更精细的模型对结果进行重排序提升Top结果的相关性。通过将系统化的提示词工程、模块化的LangGraph工作流、精准的RAG检索以及自主的智能体决策相结合你可以构建出强大、可靠且可维护的生成式AI应用。从本文的最小可行原型出发通过引入更复杂的工具、更精细的状态管理和更健壮的错误处理你的智能体将能够应对真实世界中各种复杂的任务挑战。