在实际的大模型应用开发中，Prompt Engineering（提示词工程）是连接人类意图与模型能力的核心桥梁。它远不止是“如何提问”，而是一套系统化的方法，用于设计、优化和评估引导大语言模型（LLM）完成特定任务的指令、上下文和约束。随着应用从简单的问答走向复杂的多步骤工作流，开发者需要更强大的框架来编排LLM、工具、记忆和决策逻辑。LangGraph作为LangChain生态中用于构建有状态、多参与者工作流的新星，与RAG（检索增强生成）和Agent（智能体）技术结合，正在成为构建下一代生成式AI应用的事实标准。

本文旨在为有一定Python和LLM基础的开发者提供一个从理论到实践的深度指南。我们将首先厘清Prompt Engineering的核心原则，然后以LangGraph为框架，逐步构建一个具备长期记忆、工具调用和条件分支能力的智能体，并集成RAG系统来增强其知识库。最终，你将获得一个可运行、可扩展的智能体原型，理解其内部状态流转，并掌握排查常见问题与优化性能的关键技巧。

1. 理解提示词工程：超越简单问答的指令设计

提示词工程的目标是最大化LLM在特定任务上的性能、可靠性和可控性。它不是一个静态的模板，而是一个动态的优化过程。

1.1 核心要素与设计模式

一个高效的提示词通常包含以下几个结构化部分：

1. 角色与任务定义：明确指定模型在对话中扮演的角色（如“资深Python开发顾问”）和需要完成的具体任务。
2. 上下文与背景信息：提供完成任务所需的必要信息，这可以包括用户输入、从数据库或向量库检索到的相关文档、历史对话记录等。
3. 指令与步骤：清晰、无歧义地列出模型需要遵循的步骤。对于复杂任务，分步指令比单一复杂指令更有效。
4. 输出格式约束：明确规定模型输出的格式，例如JSON、Markdown、特定结构的文本，甚至直接是代码。这极大地方便了后续的程序化处理。
5. 示例（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 wheel

2.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 pyvis

2.3 配置API密钥与环境变量

为了安全地管理API密钥，使用.env文件。

1. 在项目根目录创建.env文件。
2. 填入你的API密钥（以OpenAI为例）：

# .env OPENAI_API_KEY=sk-your-openai-api-key-here # 如需其他模型，可添加如 ANTHROPIC_API_KEY, GROQ_API_KEY 等

3. 在Python代码中通过dotenv加载：

from dotenv import load_dotenv import os load_dotenv() openai_api_key = os.getenv("OPENAI_API_KEY")

2.4 工具版本兼容性说明

不同版本库的API可能有变化。以下是撰写本文时测试兼容的版本组合，可作为参考：

注意：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: str

TypedDict提供了类型提示。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(model="gpt-4-turbo-preview", temperature=0) embeddings = OpenAIEmbeddings(model="text-embedding-3-small") # 假设我们已经有一个已加载文档的向量库 vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embeddings) 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(content=question)] # 调用绑定了工具的LLM response = llm_with_tools.invoke(messages) # 初始化返回的更新字段 updates = {"reasoning": state["reasoning"] + [f"LLM响应类型: {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(content=str(result), tool_call_id=tc["id"])) except Exception as e: tool_messages.append(ToolMessage(content=f"Error: {e}", tool_call_id=tc["id"])) else: tool_messages.append(ToolMessage(content=f"Tool {tool_name} not found.", tool_call_id=tc["id"])) # 更新状态：记录工具执行结果，并决定下一步是返回给Agent继续思考 updates = { "tool_results": tool_messages, "next_step": "agent" # 执行完工具后，返回智能体节点进行下一步决策 } return updates

4.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(content=f"[检索到的背景信息]：{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(content=response.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智能体时，以下问题是高频出现的：

6.3 性能与生产环境最佳实践

1. 异步支持：LangGraph支持异步节点。对于IO密集型操作（如网络请求、数据库查询），使用async def定义节点函数，并使用ainvoke运行图，可以显著提升吞吐量。

   async def async_retrieve_node(state: State): # 异步检索操作 docs = await retriever.ainvoke(state["question"]) return {"docs": docs}

2. 持久化检查点：使用MemorySaver或SqliteSaver等检查点存储器，可以暂停和恢复长时间运行的工作流，这对于处理复杂、多轮交互的任务至关重要。

   from langgraph.checkpoint import MemorySaver memory = MemorySaver() app = workflow.compile(checkpointer=memory) # 可以通过 thread_id 来管理不同的会话流 config = {"configurable": {"thread_id": "user_123"}} app.invoke(initial_state, config=config)

3. 可观测性与监控：集成LangSmith可以追踪每次图的执行、每个节点的输入输出、耗时和Token使用情况，便于调试和优化。

   import os os.environ["LANGCHAIN_TRACING_V2"] = "true" os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key" # 调用将被自动记录到LangSmith

4. 提示词优化与版本控制：将提示词模板外置到配置文件或数据库中，便于A/B测试和迭代更新。避免将长篇提示词硬编码在代码中。

5. 错误处理与降级：在节点函数内部使用try...except包裹核心逻辑，并更新状态以反映错误。可以设计一个专门的“错误处理”节点来接管出错的状态，提供降级响应。

6. 向量库优化：

   • 分块策略：根据文档类型调整分块大小（如技术文档256-512词，小说1024词）和重叠区域（10-20%）。
   • 元数据过滤：在检索时利用元数据（如文档来源、章节、日期）进行过滤，提高精度。
   • 重排序：在初步检索（召回）后，使用一个更精细的模型对结果进行重排序，提升Top结果的相关性。

通过将系统化的提示词工程、模块化的LangGraph工作流、精准的RAG检索以及自主的智能体决策相结合，你可以构建出强大、可靠且可维护的生成式AI应用。从本文的最小可行原型出发，通过引入更复杂的工具、更精细的状态管理和更健壮的错误处理，你的智能体将能够应对真实世界中各种复杂的任务挑战。