如果你正在学习大语言模型应用开发可能会遇到这样的困惑为什么我写的提示词Prompt效果总是不稳定为什么我的RAG系统回答得似是而非为什么Agent项目写着写着就变成了“if-else地狱”很多人以为只要把Prompt写得更长、更详细或者把最新的LangGraph、RAG框架拿来用就能做出聪明的AI应用。但吴恩达在最新的《提示工程》课程中揭示了一个核心真相构建可靠AI应用的关键不在于使用最炫酷的工具而在于掌握一套系统化的、可工程化的思维框架。工具是“术”而工程化思维是“道”。本文将基于吴恩达课程的核心思想结合LangGraph、RAG、Agent等热门技术为你拆解一套从提示词到智能体的完整开发路径。你不会只看到零散的代码片段而是会理解如何系统化地设计、评估和迭代提示词告别“玄学调参”。如何用LangGraph构建可控、可观测的复杂Agent工作流而不仅仅是串联几个LLM调用。如何让RAG系统真正“理解”你的问题而不仅仅是做关键词匹配。如何将这些技术组合成一个健壮的、可维护的AI应用。我们将通过一个完整的“技术文档问答与摘要生成Agent”项目来贯穿这些概念并提供可直接运行的代码。无论你是想入门AI应用开发还是希望提升现有项目的工程化水平这篇文章都将提供清晰的路线图和实战指南。1. 重新理解提示工程从“咒语”到“可测试的代码”很多人把写Prompt看作是一种“魔法咒语”效果好坏全凭运气和灵感。吴恩达课程的核心转变在于将提示工程视为软件工程的一部分。这意味着Prompt也需要被设计、测试、版本控制和迭代。1.1 提示词的三个核心层级一个工程化的提示词通常包含以下结构这远比堆砌指令更有效系统指令System Instruction定义AI的“角色”和基本行为准则。这是最稳定、最少修改的部分。任务上下文与格式Task Context Format提供本次任务的具体背景信息并明确要求输出的格式如JSON、Markdown。少样本示例Few-shot Examples提供1-3个高质量的输入-输出示例这是引导模型理解复杂任务最有效的方式之一。错误示例 vs 工程化示例# 错误示例模糊、冗长、难以评估 prompt “请分析一下这段代码告诉我有什么问题并给出优化建议。” # 工程化示例角色清晰、任务具体、格式明确、有示例 system_instruction “你是一位经验丰富的Python代码审查专家。你的职责是以清晰、专业的方式指出代码中的问题并提供可操作的改进建议。” task_context “”” 请审查以下Python函数。你的分析需要包含 1. **潜在问题**列出所有发现的代码缺陷、性能瓶颈或不良实践。 2. **严重等级**对每个问题标注【高危】、【中危】或【低危】。 3. **优化建议**为每个问题提供具体的修改代码示例。 4. **总结**用一句话总结代码的整体质量。 请严格按照以下JSON格式输出 { “issues”: [ { “description”: “问题描述”, “severity”: “高危/中危/低危”, “suggestion”: “修改建议和示例代码” } ], “summary”: “整体质量总结” } “”” few_shot_example “”” 示例输入函数 def process_data(lst): return [i*2 for i in lst] 示例输出 { “issues”: [ { “description”: “函数名‘process_data’过于泛化未能体现其‘元素翻倍’的具体功能。”, “severity”: “低危”, “suggestion”: “建议重命名为‘double_elements’例如def double_elements(lst): return [i*2 for i in lst]” } ], “summary”: “函数功能简单且正确但命名可读性有待提高。” } “”” # 最终组合提示词 engineered_prompt f“{system_instruction}\n\n{task_context}\n\n{few_shot_example}\n\n现在请审查这个函数{user_code}”1.2 提示词的评估与迭代工程化的关键在于可测量。你需要为关键任务设计评估标准。简单任务可以使用精确匹配Exact Match或关键词检查。复杂任务需要设计模型评估LLM-as-a-Judge。即用另一个LLM通常是更强大的模型根据一套评分规则来评估输出质量。# 一个简单的模型评估示例使用OpenAI API import openai def evaluate_response_with_llm(question, ideal_answer, model_answer, criteria): “”” 使用LLM作为裁判来评估模型回答质量。 criteria: 评估标准列表如 [‘准确性’ ‘完整性’ ‘清晰度’] “”” evaluation_prompt f“”” 你是一位公正的评估员。请根据以下标准评估“模型回答”相对于“问题”和“参考答案”的质量。 评估标准{‘ ‘.join(criteria)} 问题{question} 参考答案{ideal_answer} 模型回答{model_answer} 请为每个标准打分1-5分5为最佳并给出简要理由。以JSON格式输出 {{ “scores”: {{“准确性”: 分数 “完整性”: 分数 “清晰度”: 分数}}, “reason”: “总体评价理由” }} “”” response openai.chat.completions.create( model“gpt-4” messages[{“role”: “user” “content”: evaluation_prompt}], temperature0 ) return response.choices[0].message.content通过批量测试和评估你可以量化不同Prompt版本的效果从而进行有依据的优化而不是盲目猜测。2. LangGraph将AI工作流从“流程图”变为“状态机”当你需要构建一个涉及多步骤、有条件分支、有状态记忆的复杂Agent时简单的链式调用LangChain Expression Language很快就会变得难以维护。这就是LangGraph要解决的问题。2.1 核心概念图Graph与状态State图Graph由节点Nodes和边Edges组成。节点代表一个执行单元如调用LLM、执行工具边定义了节点之间的流转逻辑。状态State一个贯穿整个工作流的共享数据字典。每个节点都可以读取和修改状态。这是LangGraph与简单链最本质的区别。想象一个客服Agent的工作流节点A理解用户意图。根据意图决定走边1查询知识库或边2转人工。节点B若走边1则执行RAG检索。节点C组织答案并回复。 这个动态的、有状态的过程用LangGraph可以很清晰地建模。2.2 一个简单的LangGraph智能体示例我们来构建一个具有自我反思能力的问答Agent。它的工作流是先尝试直接回答如果自己不确定就调用搜索工具查找资料最后整合信息回复。# 文件self_reflective_agent.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_community.tools import TavilySearchResults # 1. 定义状态结构State Schema class AgentState(TypedDict): “””定义贯穿工作流的状态字典。“”” question: str # 用户问题 initial_answer: str # 初始答案 confidence: float # 置信度 (0-1) search_results: List[str] # 搜索结果 final_answer: str # 最终答案 # 2. 初始化模型和工具 llm ChatOpenAI(model“gpt-3.5-turbo”) search_tool TavilySearchResults() # 需要TAVILY_API_KEY # 3. 定义各个节点Nodes def generate_initial_answer(state: AgentState): “””节点A尝试直接回答问题并评估自己的置信度。“”” prompt f“请直接回答以下问题{state[‘question’]}。在答案末尾请单独用一行输出你的置信度格式为‘置信度0.XX’。” response llm.invoke(prompt) content response.content # 简单解析答案和置信度 if “置信度” in content: answer_part, conf_part content.rsplit(“置信度” 1) try: confidence float(conf_part.strip()) except: confidence 0.5 answer answer_part.strip() else: answer content confidence 0.5 # 更新状态 return {“initial_answer”: answer “confidence”: confidence} def decide_to_search(state: AgentState): “””路由节点根据置信度决定下一步。“”” # 如果置信度低就去搜索 if state[‘confidence’] 0.7: return “search” else: return “finalize” def search_for_info(state: AgentState): “””节点B调用搜索工具获取更多信息。“”” results search_tool.invoke(state[‘question’]) # 假设我们取前3条结果的内容 search_info [r[‘content’] for r in results[:3]] return {“search_results”: search_info} def synthesize_final_answer(state: AgentState): “””节点C综合初始答案和搜索结果生成最终答案。“”” context state[‘initial_answer’] if state[‘search_results’]: context “\n\n补充信息\n” “\n”.join(state[‘search_results’]) prompt f“基于以下信息请重新组织一个准确、完整的答案来回答问题{state[‘question’]}\n\n信息{context}” response llm.invoke(prompt) return {“final_answer”: response.content} def direct_final_answer(state: AgentState): “””节点D当置信度高时直接使用初始答案作为最终答案。“”” return {“final_answer”: state[‘initial_answer’]} # 4. 构建图Graph workflow StateGraph(AgentState) # 添加节点 workflow.add_node(“generate” generate_initial_answer) workflow.add_node(“search” search_for_info) workflow.add_node(“synthesize” synthesize_final_answer) workflow.add_node(“finalize_direct” direct_final_answer) # 设置入口 workflow.set_entry_point(“generate”) # 添加边路由 workflow.add_conditional_edges( “generate” decide_to_search, # 路由判断函数 { “search”: “search” # 如果返回“search”则前往“search”节点 “finalize”: “finalize_direct” # 如果返回“finalize”则前往“finalize_direct”节点 } ) workflow.add_edge(“search” “synthesize”) # 搜索完成后必然去合成答案 workflow.add_edge(“synthesize” END) # 合成答案后结束 workflow.add_edge(“finalize_direct” END) # 直接最终化后结束 # 编译图 app workflow.compile() # 5. 运行工作流 if __name__ “__main__”: # 初始化状态 initial_state: AgentState {“question”: “LangGraph和LangChain的主要区别是什么”} # 执行图 final_state app.invoke(initial_state) print(“最终答案” final_state[“final_answer”]) print(“\n完整执行轨迹可选:” final_state[“__pregel_messages”]) # LangGraph内置的调试信息这个示例展示了LangGraph的核心价值将复杂的逻辑判断是否搜索内化为图的路由使代码结构清晰且易于扩展。你可以通过可视化工具查看这个图的结构直观理解数据流。3. RAG的工程化实践超越简单的向量检索RAG检索增强生成的核心问题不是“如何实现”而是“如何实现得好”。一个基础的RAG系统很容易搭建但要让其回答准确、相关需要一系列工程化改进。3.1 RAG的三大常见痛点与解决方案痛点现象工程化解决方案检索不相关返回的文档片段与问题无关导致生成答案胡编乱造。1. 查询优化使用LLM对原始问题进行重写、扩展或生成假设性答案用于检索。2. 优质分块根据文档类型代码、论文、手册采用不同的分块策略和大小。3. 重排序Re-ranking使用专门的重排序模型如Cohere Rerank BGE Reranker对初步检索结果进行精排。信息不完整答案只覆盖了部分相关文档遗漏关键点。1. 多向量检索对同一文档生成摘要、关键词、问题等多种向量表示进行检索。2. 递归检索先检索出粗粒度文档再对其内容进行更细粒度的二次检索。3. 融合检索结合关键词检索如BM25和向量检索如Embedding的结果。生成不遵从模型无视检索到的上下文基于自身知识胡编乱造。1. 强化提示在系统指令中明确要求“严格基于提供的上下文”。2. 引用溯源要求模型在生成答案时注明引用的文档片段编号或位置。3. 后处理验证用另一个LLM检查生成答案是否与提供上下文一致。3.2 实现一个带重排序的RAG管道下面是一个使用LangChain和BGE重排序模型的高级RAG示例。# 文件advanced_rag_pipeline.py from langchain_chroma import Chroma from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import LLMChainExtractor # 假设我们使用FlagEmbedding的BGE重排序器 from FlagEmbedding import FlagReranker # 1. 加载和分块文档 loader TextLoader(“./technical_manual.txt”) documents loader.load() text_splitter RecursiveCharacterTextSplitter(chunk_size500, chunk_overlap50) chunks text_splitter.split_documents(documents) # 2. 创建向量数据库 embeddings OpenAIEmbeddings() # 或使用本地模型如 BGE-M3 vectorstore Chroma.from_documents(documentschunks, embeddingembeddings, persist_directory“./chroma_db”) base_retriever vectorstore.as_retriever(search_kwargs{“k”: 10}) # 初步检索10个片段 # 3. 自定义重排序器 class BGEReranker: def __init__(self, model_name“BAAI/bge-reranker-large”): self.reranker FlagReranker(model_name, use_fp16True) # 加载模型 def compress_documents(self, query: str, documents: List[Document]): “””对检索到的文档进行重排序。“”” if not documents: return [] # 准备query document对 pairs [(query, doc.page_content) for doc in documents] # 计算相关性分数 scores self.reranker.compute_score(pairs) # 根据分数排序 scored_docs list(zip(scores, documents)) scored_docs.sort(keylambda x: x[0], reverseTrue) # 返回Top-K个文档例如前3个 return [doc for score, doc in scored_docs[:3]] # 4. 创建带重排序的检索器 reranker BGEReranker() compression_retriever ContextualCompressionRetriever( base_compressorreranker, base_retrieverbase_retriever ) # 5. 构建RAG链 llm ChatOpenAI(model“gpt-4”) from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough # 提示词模板强制模型引用上下文 template “””你是一个技术文档助手。请严格根据以下提供的上下文来回答问题。如果上下文不包含答案请直接说“根据提供的资料我无法回答这个问题”。在答案中请尽可能引用上下文中的内容。 上下文{context} 问题{question} 请给出答案””” prompt ChatPromptTemplate.from_template(template) rag_chain ( {“context”: compression_retriever “question”: RunnablePassthrough()} | prompt | llm | StrOutputParser() ) # 6. 提问 question “如何在LangGraph中实现条件分支” answer rag_chain.invoke(question) print(“问题” question) print(“答案\n” answer)这个管道通过重排序筛选出最相关的3个片段并利用强约束的提示词显著提升了RAG回答的准确性和可靠性。4. 构建智能体AgentLangGraph RAG 工具调用现在我们将前面所学组合起来构建一个更强大的智能体它能根据问题类型自动选择是使用内部知识库RAG回答还是使用搜索引擎工具获取最新信息并能将多步结果进行汇总。4.1 智能体架构设计我们将构建一个具有以下能力的智能体问题分类判断问题是关于内部技术文档还是需要外部实时信息。路由执行内部问题 → 调用RAG检索链。外部问题 → 调用搜索工具。复杂问题 → 可能先搜索再结合RAG结果进行综合。答案合成将检索到的信息组织成连贯、友好的答案。4.2 完整代码实现# 文件tech_doc_agent.py import os from typing import TypedDict, List, Literal from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_chroma import Chroma from langchain_core.prompts import ChatPromptTemplate from langchain_community.tools import TavilySearchResults from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough # ------------------ 0. 状态定义 ------------------ class AgentState(TypedDict): “””智能体的全局状态。“”” user_query: str query_type: Literal[“internal” “external” “hybrid”] | None # 问题类型 internal_context: List[str] # RAG检索到的内部文档 external_context: List[str] # 搜索到的外部信息 final_answer: str # ------------------ 1. 初始化组件 ------------------ llm ChatOpenAI(model“gpt-3.5-turbo”) search_tool TavilySearchResults(max_results3) embeddings OpenAIEmbeddings() # 假设已有一个构建好的内部知识库向量库 vectorstore Chroma(persist_directory“./chroma_db” embedding_functionembeddings) retriever vectorstore.as_retriever(search_kwargs{“k”: 4}) # ------------------ 2. 定义各类提示词模板 ------------------ classification_prompt ChatPromptTemplate.from_template(“”” 请将用户问题分类为以下三种类型之一 - **internal**: 问题明确关于我们内部的技术栈、文档或代码如LangGraph RAG 我们的API规范。 - **external**: 问题需要最新的、实时的外部信息或通用知识如今天的天气 某公司最新新闻。 - **hybrid**: 问题可能既需要内部知识背景也需要外部最新信息补充。 只输出一个单词internal external 或 hybrid。 用户问题{question} 分类结果”””) rag_prompt ChatPromptTemplate.from_template(“”” 你是一个内部技术专家。请严格根据以下内部文档片段来回答问题。 内部文档 {internal_context} 问题{question} 如果文档中没有相关信息请明确说明“内部文档未涵盖此点”。请基于文档给出答案”””) synthesis_prompt ChatPromptTemplate.from_template(“”” 你是一个信息整合助手。用户提出了一个问题我们已经收集了以下信息 用户问题{question} 【内部技术资料】 {internal_context} 【外部网络信息】 {external_context} 请综合以上所有信息生成一个全面、准确、结构清晰的最终答案。如果信息间有冲突以内部技术资料为准。如果某些方面缺乏信息请如实说明。 最终答案”””) # ------------------ 3. 定义图节点 ------------------ def classify_query(state: AgentState): “””节点A问题分类。“”” chain classification_prompt | llm | StrOutputParser() query_type chain.invoke({“question”: state[“user_query”]}).strip().lower() # 简单清理结果 if query_type not in [“internal” “external” “hybrid”]: query_type “hybrid” # 默认混合型 return {“query_type”: query_type} def retrieve_internal_info(state: AgentState): “””节点B检索内部知识库RAG。“”” docs retriever.invoke(state[“user_query”]) context “\n---\n”.join([doc.page_content for doc in docs]) return {“internal_context”: context} def search_external_info(state: AgentState): “””节点C搜索外部信息。“”” results search_tool.invoke(state[“user_query”]) # 提取摘要或内容 external_info [f“来源{r[‘title’]}\n内容{r[‘content’]}” for r in results] return {“external_context”: “\n---\n”.join(external_info)} def route_query(state: AgentState): “””路由节点根据分类结果决定下一步流程。“”” q_type state[“query_type”] if q_type “internal”: return “internal_only” elif q_type “external”: return “external_only” else: # hybrid return “hybrid_process” def answer_with_rag(state: AgentState): “””节点D仅使用RAG回答内部问题。“”” chain rag_prompt | llm | StrOutputParser() answer chain.invoke({ “question”: state[“user_query”] “internal_context”: state[“internal_context”] }) return {“final_answer”: answer} def answer_with_search(state: AgentState): “””节点E仅使用搜索回答外部问题。“”” # 这里可以简单使用搜索结果的摘要也可以让LLM总结 summary_chain ( ChatPromptTemplate.from_template(“请根据以下网络搜索结果总结回答这个问题{question}\n\n结果{results}”) | llm | StrOutputParser() ) answer summary_chain.invoke({ “question”: state[“user_query”] “results”: state[“external_context”] }) return {“final_answer”: answer} def synthesize_hybrid_answer(state: AgentState): “””节点F综合内部和外部信息生成答案。“”” chain synthesis_prompt | llm | StrOutputParser() answer chain.invoke({ “question”: state[“user_query”] “internal_context”: state.get(“internal_context” “无内部资料”) “external_context”: state.get(“external_context” “无外部信息”) }) return {“final_answer”: answer} # ------------------ 4. 构建工作流图 ------------------ workflow StateGraph(AgentState) # 添加所有节点 workflow.add_node(“classify” classify_query) workflow.add_node(“retrieve” retrieve_internal_info) workflow.add_node(“search” search_external_info) workflow.add_node(“answer_rag” answer_with_rag) workflow.add_node(“answer_search” answer_with_search) workflow.add_node(“synthesize” synthesize_hybrid_answer) # 设置入口 workflow.set_entry_point(“classify”) # 添加边和条件路由 workflow.add_edge(“classify” “route_decision”) # 条件路由 workflow.add_conditional_edges( “route_decision” route_query { “internal_only”: “retrieve” “external_only”: “search” “hybrid_process”: [“retrieve” “search”] # 并行执行两个分支 } ) # 定义并行分支后的汇聚点 workflow.add_edge(“retrieve” “answer_rag”) workflow.add_edge(“search” “answer_search”) # 设置最终出口 workflow.add_edge(“answer_rag” END) workflow.add_edge(“answer_search” END) # 对于混合流程需要先并行执行retrieve和search再汇聚到synthesize # 这里需要用到LangGraph的“映射”和“归约”概念为简化我们假设混合流程是串行的。 # 更复杂的并行-汇聚逻辑需要定义子图此处为清晰起见我们修改为串行逻辑。 def route_after_parallel(state): # 这是一个简化处理。实际中应等待并行分支完成。 if state[“query_type”] “hybrid”: return “synthesize” else: return END # 重新构建一个简化版的图避免在示例中引入过于复杂的并行控制 # 我们改为hybrid流程也先检索再搜索最后合成。 workflow_simple StateGraph(AgentState) workflow_simple.add_node(“classify” classify_query) workflow_simple.add_node(“retrieve” retrieve_internal_info) workflow_simple.add_node(“search” search_external_info) workflow_simple.add_node(“answer_rag” answer_with_rag) workflow_simple.add_node(“answer_search” answer_with_search) workflow_simple.add_node(“synthesize” synthesize_hybrid_answer) workflow_simple.set_entry_point(“classify”) workflow_simple.add_edge(“classify” “route_decision_simple”) def route_query_simple(state): q_type state[“query_type”] if q_type “internal”: return “retrieve” elif q_type “external”: return “search” else: # hybrid return “retrieve_for_hybrid” workflow_simple.add_conditional_edges( “route_decision_simple” route_query_simple { “retrieve”: “retrieve” “search”: “search” “retrieve_for_hybrid”: “retrieve” } ) workflow_simple.add_edge(“retrieve” “after_retrieve”) def route_after_retrieve(state): if state[“query_type”] “internal”: return “answer_rag” else: # hybrid return “search_for_hybrid” # hybrid流程继续搜索 workflow_simple.add_conditional_edges(“after_retrieve” route_after_retrieve {“answer_rag”: “answer_rag” “search_for_hybrid”: “search”}) workflow_simple.add_edge(“search” “after_search”) def route_after_search(state): if state[“query_type”] “external”: return “answer_search” else: # hybrid return “synthesize” workflow_simple.add_conditional_edges(“after_search” route_after_search {“answer_search”: “answer_search” “synthesize”: “synthesize”}) workflow_simple.add_edge(“answer_rag” END) workflow_simple.add_edge(“answer_search” END) workflow_simple.add_edge(“synthesize” END) # 编译图 app workflow_simple.compile() # ------------------ 5. 运行智能体 ------------------ if __name__ “__main__”: test_queries [ “LangGraph中的状态State如何定义” # 内部问题 “今天OpenAI有什么最新动态” # 外部问题 “对比一下LangGraph和AutoGen的最新版本特性。” # 混合问题 ] for query in test_queries: print(f“\n{‘’*50}”) print(f“用户问题{query}”) initial_state: AgentState {“user_query”: query} result app.invoke(initial_state) print(f“最终答案\n{result[‘final_answer’]}”) print(f“{‘’*50}”)这个智能体展示了如何将分类、路由、工具调用搜索、RAG检索、信息合成等多个步骤通过LangGraph组织成一个清晰、可控的工作流。你可以通过LangGraph Studio可视化这个工作流直观地调试每个节点的输入输出。5. 环境准备、运行与问题排查5.1 环境准备确保你已安装Python 3.9并创建虚拟环境。# 1. 创建并激活虚拟环境 python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 2. 安装核心库 pip install langgraph langchain langchain-openai langchain-chroma langchain-community pip install openai tiktoken chromadb # 3. 安装可选组件用于搜索和重排序 pip install tavily-python # 搜索工具需要API Key # pip install FlagEmbedding # 如需本地重排序模型5.2 设置API密钥在代码运行前需要设置必要的环境变量。# 在终端中设置或写在 .env 文件中 export OPENAI_API_KEY“sk-...” # 你的OpenAI API Key export TAVILY_API_KEY“tvly-...” # 你的Tavily API Key用于搜索或者在Python代码中设置import os os.environ[“OPENAI_API_KEY”] “sk-...”5.3 运行示例代码将前面的代码示例保存为独立的.py文件。确保你有technical_manual.txt文档用于构建RAG知识库或修改代码加载你自己的文档。按顺序运行# 首先构建向量数据库运行一次即可 python advanced_rag_pipeline.py # 这会创建./chroma_db目录 # 然后运行智能体 python tech_doc_agent.py5.4 常见问题排查表问题现象可能原因排查步骤ModuleNotFoundError依赖未正确安装。1. 确认虚拟环境已激活。2. 运行 pip listAuthenticationError/Invalid API KeyAPI密钥错误或未设置。1. 检查os.environ[‘OPENAI_API_KEY’]是否已正确赋值。2. 确认密钥有效且有余额。3. 对于Tavily等工具检查是否已注册并获取API Key。向量数据库连接失败persist_directory路径错误或权限不足。1. 确认./chroma_db目录存在或可创建。2. 尝试使用绝对路径。RAG检索结果不相关文档分块策略不佳或Embedding模型不合适。1. 调整chunk_size和chunk_overlap。2. 尝试不同的文本分割器如按标题分割。3. 考虑更换或微调Embedding模型。LangGraph图编译或运行错误状态State定义与节点返回值不匹配。1. 检查AgentState的字段类型是否与每个节点返回的字典键完全匹配。2. 使用app.get_graph().draw_mermaid()输出图结构检查逻辑。3. 在节点函数内添加print(state)调试状态变化。Agent回答质量差提示词设计不佳或LLM模型能力不足。1. 优化系统指令和任务描述使其更清晰、更具约束力。2. 增加少样本示例Few-shot Examples。3. 升级到更强大的LLM如GPT-4。4. 引入评估环节量化优化效果。6. 最佳实践与工程建议提示词版本管理像管理代码一样管理你的提示词。使用配置文件如YAML、JSON或专门的提示词管理平台来存储不同版本的Prompt便于A/B测试和回滚。图的可观测性利用LangGraph的内置追踪和LangSmith等工具记录每个节点的输入输出、耗时和异常。这对于调试复杂工作流至关重要。优雅降级与超时控制在调用外部API如LLM、搜索时务必设置超时和重试机制并为失败情况设计备用路径如返回缓存结果、使用更简单的本地模型。测试驱动开发为你的Agent工作流编写单元测试和集成测试。测试应包括正常用例、边界用例、错误处理如网络超时、API限流。安全性输入净化对用户输入进行过滤防止Prompt注入攻击。输出审查对AI生成的内容进行安全检查避免输出有害或不适当信息。权限控制确保工具如文件读写、数据库访问在最小必要权限下运行。成本控制监控LLM的Token消耗。对于非关键路径考虑使用更经济的模型。对RAG检索结果进行摘要或过滤减少送入LLM的上下文长度。从“咒语式”的Prompt尝试到基于状态和图的可控Agent工作流再到工程化的RAG管道构建可靠AI应用的本质是将不确定性封装在可测试、可观测、可迭代的模块中。吴恩达课程的精髓正在于此它提供了一套方法论而非零散的技巧。本文通过一个从简到繁的示例项目串联了Prompt Engineering、LangGraph、RAG和Agent的核心概念与实践。你可以以此为起点深入LangGraph探索子图、检查点持久化状态、人工审批节点等高级特性构建更稳定、支持长对话的Agent。优化RAG尝试不同的Embedding模型、重排序器、查询转换策略甚至用LLM来评估检索质量持续提升知识库的召回率和准确率。设计多Agent系统让多个具备不同技能的Agent协作用LangGraph来编排它们之间的通信和任务分配。真正的挑战不在于理解单个工具而在于如何将它们有机组合解决真实的业务问题。现在你可以尝试用这套框架去改造你手头那个“if-else地狱”的原型或者开始构建一个全新的、真正智能的AI应用了。
如果你正在学习大语言模型应用开发可能会遇到这样的困惑为什么我写的提示词Prompt效果总是不稳定为什么我的RAG系统回答得似是而非为什么Agent项目写着写着就变成了“if-else地狱”很多人以为只要把Prompt写得更长、更详细或者把最新的LangGraph、RAG框架拿来用就能做出聪明的AI应用。但吴恩达在最新的《提示工程》课程中揭示了一个核心真相构建可靠AI应用的关键不在于使用最炫酷的工具而在于掌握一套系统化的、可工程化的思维框架。工具是“术”而工程化思维是“道”。本文将基于吴恩达课程的核心思想结合LangGraph、RAG、Agent等热门技术为你拆解一套从提示词到智能体的完整开发路径。你不会只看到零散的代码片段而是会理解如何系统化地设计、评估和迭代提示词告别“玄学调参”。如何用LangGraph构建可控、可观测的复杂Agent工作流而不仅仅是串联几个LLM调用。如何让RAG系统真正“理解”你的问题而不仅仅是做关键词匹配。如何将这些技术组合成一个健壮的、可维护的AI应用。我们将通过一个完整的“技术文档问答与摘要生成Agent”项目来贯穿这些概念并提供可直接运行的代码。无论你是想入门AI应用开发还是希望提升现有项目的工程化水平这篇文章都将提供清晰的路线图和实战指南。1. 重新理解提示工程从“咒语”到“可测试的代码”很多人把写Prompt看作是一种“魔法咒语”效果好坏全凭运气和灵感。吴恩达课程的核心转变在于将提示工程视为软件工程的一部分。这意味着Prompt也需要被设计、测试、版本控制和迭代。1.1 提示词的三个核心层级一个工程化的提示词通常包含以下结构这远比堆砌指令更有效系统指令System Instruction定义AI的“角色”和基本行为准则。这是最稳定、最少修改的部分。任务上下文与格式Task Context Format提供本次任务的具体背景信息并明确要求输出的格式如JSON、Markdown。少样本示例Few-shot Examples提供1-3个高质量的输入-输出示例这是引导模型理解复杂任务最有效的方式之一。错误示例 vs 工程化示例# 错误示例模糊、冗长、难以评估 prompt “请分析一下这段代码告诉我有什么问题并给出优化建议。” # 工程化示例角色清晰、任务具体、格式明确、有示例 system_instruction “你是一位经验丰富的Python代码审查专家。你的职责是以清晰、专业的方式指出代码中的问题并提供可操作的改进建议。” task_context “”” 请审查以下Python函数。你的分析需要包含 1. **潜在问题**列出所有发现的代码缺陷、性能瓶颈或不良实践。 2. **严重等级**对每个问题标注【高危】、【中危】或【低危】。 3. **优化建议**为每个问题提供具体的修改代码示例。 4. **总结**用一句话总结代码的整体质量。 请严格按照以下JSON格式输出 { “issues”: [ { “description”: “问题描述”, “severity”: “高危/中危/低危”, “suggestion”: “修改建议和示例代码” } ], “summary”: “整体质量总结” } “”” few_shot_example “”” 示例输入函数 def process_data(lst): return [i*2 for i in lst] 示例输出 { “issues”: [ { “description”: “函数名‘process_data’过于泛化未能体现其‘元素翻倍’的具体功能。”, “severity”: “低危”, “suggestion”: “建议重命名为‘double_elements’例如def double_elements(lst): return [i*2 for i in lst]” } ], “summary”: “函数功能简单且正确但命名可读性有待提高。” } “”” # 最终组合提示词 engineered_prompt f“{system_instruction}\n\n{task_context}\n\n{few_shot_example}\n\n现在请审查这个函数{user_code}”1.2 提示词的评估与迭代工程化的关键在于可测量。你需要为关键任务设计评估标准。简单任务可以使用精确匹配Exact Match或关键词检查。复杂任务需要设计模型评估LLM-as-a-Judge。即用另一个LLM通常是更强大的模型根据一套评分规则来评估输出质量。# 一个简单的模型评估示例使用OpenAI API import openai def evaluate_response_with_llm(question, ideal_answer, model_answer, criteria): “”” 使用LLM作为裁判来评估模型回答质量。 criteria: 评估标准列表如 [‘准确性’ ‘完整性’ ‘清晰度’] “”” evaluation_prompt f“”” 你是一位公正的评估员。请根据以下标准评估“模型回答”相对于“问题”和“参考答案”的质量。 评估标准{‘ ‘.join(criteria)} 问题{question} 参考答案{ideal_answer} 模型回答{model_answer} 请为每个标准打分1-5分5为最佳并给出简要理由。以JSON格式输出 {{ “scores”: {{“准确性”: 分数 “完整性”: 分数 “清晰度”: 分数}}, “reason”: “总体评价理由” }} “”” response openai.chat.completions.create( model“gpt-4” messages[{“role”: “user” “content”: evaluation_prompt}], temperature0 ) return response.choices[0].message.content通过批量测试和评估你可以量化不同Prompt版本的效果从而进行有依据的优化而不是盲目猜测。2. LangGraph将AI工作流从“流程图”变为“状态机”当你需要构建一个涉及多步骤、有条件分支、有状态记忆的复杂Agent时简单的链式调用LangChain Expression Language很快就会变得难以维护。这就是LangGraph要解决的问题。2.1 核心概念图Graph与状态State图Graph由节点Nodes和边Edges组成。节点代表一个执行单元如调用LLM、执行工具边定义了节点之间的流转逻辑。状态State一个贯穿整个工作流的共享数据字典。每个节点都可以读取和修改状态。这是LangGraph与简单链最本质的区别。想象一个客服Agent的工作流节点A理解用户意图。根据意图决定走边1查询知识库或边2转人工。节点B若走边1则执行RAG检索。节点C组织答案并回复。 这个动态的、有状态的过程用LangGraph可以很清晰地建模。2.2 一个简单的LangGraph智能体示例我们来构建一个具有自我反思能力的问答Agent。它的工作流是先尝试直接回答如果自己不确定就调用搜索工具查找资料最后整合信息回复。# 文件self_reflective_agent.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_community.tools import TavilySearchResults # 1. 定义状态结构State Schema class AgentState(TypedDict): “””定义贯穿工作流的状态字典。“”” question: str # 用户问题 initial_answer: str # 初始答案 confidence: float # 置信度 (0-1) search_results: List[str] # 搜索结果 final_answer: str # 最终答案 # 2. 初始化模型和工具 llm ChatOpenAI(model“gpt-3.5-turbo”) search_tool TavilySearchResults() # 需要TAVILY_API_KEY # 3. 定义各个节点Nodes def generate_initial_answer(state: AgentState): “””节点A尝试直接回答问题并评估自己的置信度。“”” prompt f“请直接回答以下问题{state[‘question’]}。在答案末尾请单独用一行输出你的置信度格式为‘置信度0.XX’。” response llm.invoke(prompt) content response.content # 简单解析答案和置信度 if “置信度” in content: answer_part, conf_part content.rsplit(“置信度” 1) try: confidence float(conf_part.strip()) except: confidence 0.5 answer answer_part.strip() else: answer content confidence 0.5 # 更新状态 return {“initial_answer”: answer “confidence”: confidence} def decide_to_search(state: AgentState): “””路由节点根据置信度决定下一步。“”” # 如果置信度低就去搜索 if state[‘confidence’] 0.7: return “search” else: return “finalize” def search_for_info(state: AgentState): “””节点B调用搜索工具获取更多信息。“”” results search_tool.invoke(state[‘question’]) # 假设我们取前3条结果的内容 search_info [r[‘content’] for r in results[:3]] return {“search_results”: search_info} def synthesize_final_answer(state: AgentState): “””节点C综合初始答案和搜索结果生成最终答案。“”” context state[‘initial_answer’] if state[‘search_results’]: context “\n\n补充信息\n” “\n”.join(state[‘search_results’]) prompt f“基于以下信息请重新组织一个准确、完整的答案来回答问题{state[‘question’]}\n\n信息{context}” response llm.invoke(prompt) return {“final_answer”: response.content} def direct_final_answer(state: AgentState): “””节点D当置信度高时直接使用初始答案作为最终答案。“”” return {“final_answer”: state[‘initial_answer’]} # 4. 构建图Graph workflow StateGraph(AgentState) # 添加节点 workflow.add_node(“generate” generate_initial_answer) workflow.add_node(“search” search_for_info) workflow.add_node(“synthesize” synthesize_final_answer) workflow.add_node(“finalize_direct” direct_final_answer) # 设置入口 workflow.set_entry_point(“generate”) # 添加边路由 workflow.add_conditional_edges( “generate” decide_to_search, # 路由判断函数 { “search”: “search” # 如果返回“search”则前往“search”节点 “finalize”: “finalize_direct” # 如果返回“finalize”则前往“finalize_direct”节点 } ) workflow.add_edge(“search” “synthesize”) # 搜索完成后必然去合成答案 workflow.add_edge(“synthesize” END) # 合成答案后结束 workflow.add_edge(“finalize_direct” END) # 直接最终化后结束 # 编译图 app workflow.compile() # 5. 运行工作流 if __name__ “__main__”: # 初始化状态 initial_state: AgentState {“question”: “LangGraph和LangChain的主要区别是什么”} # 执行图 final_state app.invoke(initial_state) print(“最终答案” final_state[“final_answer”]) print(“\n完整执行轨迹可选:” final_state[“__pregel_messages”]) # LangGraph内置的调试信息这个示例展示了LangGraph的核心价值将复杂的逻辑判断是否搜索内化为图的路由使代码结构清晰且易于扩展。你可以通过可视化工具查看这个图的结构直观理解数据流。3. RAG的工程化实践超越简单的向量检索RAG检索增强生成的核心问题不是“如何实现”而是“如何实现得好”。一个基础的RAG系统很容易搭建但要让其回答准确、相关需要一系列工程化改进。3.1 RAG的三大常见痛点与解决方案痛点现象工程化解决方案检索不相关返回的文档片段与问题无关导致生成答案胡编乱造。1. 查询优化使用LLM对原始问题进行重写、扩展或生成假设性答案用于检索。2. 优质分块根据文档类型代码、论文、手册采用不同的分块策略和大小。3. 重排序Re-ranking使用专门的重排序模型如Cohere Rerank BGE Reranker对初步检索结果进行精排。信息不完整答案只覆盖了部分相关文档遗漏关键点。1. 多向量检索对同一文档生成摘要、关键词、问题等多种向量表示进行检索。2. 递归检索先检索出粗粒度文档再对其内容进行更细粒度的二次检索。3. 融合检索结合关键词检索如BM25和向量检索如Embedding的结果。生成不遵从模型无视检索到的上下文基于自身知识胡编乱造。1. 强化提示在系统指令中明确要求“严格基于提供的上下文”。2. 引用溯源要求模型在生成答案时注明引用的文档片段编号或位置。3. 后处理验证用另一个LLM检查生成答案是否与提供上下文一致。3.2 实现一个带重排序的RAG管道下面是一个使用LangChain和BGE重排序模型的高级RAG示例。# 文件advanced_rag_pipeline.py from langchain_chroma import Chroma from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import LLMChainExtractor # 假设我们使用FlagEmbedding的BGE重排序器 from FlagEmbedding import FlagReranker # 1. 加载和分块文档 loader TextLoader(“./technical_manual.txt”) documents loader.load() text_splitter RecursiveCharacterTextSplitter(chunk_size500, chunk_overlap50) chunks text_splitter.split_documents(documents) # 2. 创建向量数据库 embeddings OpenAIEmbeddings() # 或使用本地模型如 BGE-M3 vectorstore Chroma.from_documents(documentschunks, embeddingembeddings, persist_directory“./chroma_db”) base_retriever vectorstore.as_retriever(search_kwargs{“k”: 10}) # 初步检索10个片段 # 3. 自定义重排序器 class BGEReranker: def __init__(self, model_name“BAAI/bge-reranker-large”): self.reranker FlagReranker(model_name, use_fp16True) # 加载模型 def compress_documents(self, query: str, documents: List[Document]): “””对检索到的文档进行重排序。“”” if not documents: return [] # 准备query document对 pairs [(query, doc.page_content) for doc in documents] # 计算相关性分数 scores self.reranker.compute_score(pairs) # 根据分数排序 scored_docs list(zip(scores, documents)) scored_docs.sort(keylambda x: x[0], reverseTrue) # 返回Top-K个文档例如前3个 return [doc for score, doc in scored_docs[:3]] # 4. 创建带重排序的检索器 reranker BGEReranker() compression_retriever ContextualCompressionRetriever( base_compressorreranker, base_retrieverbase_retriever ) # 5. 构建RAG链 llm ChatOpenAI(model“gpt-4”) from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough # 提示词模板强制模型引用上下文 template “””你是一个技术文档助手。请严格根据以下提供的上下文来回答问题。如果上下文不包含答案请直接说“根据提供的资料我无法回答这个问题”。在答案中请尽可能引用上下文中的内容。 上下文{context} 问题{question} 请给出答案””” prompt ChatPromptTemplate.from_template(template) rag_chain ( {“context”: compression_retriever “question”: RunnablePassthrough()} | prompt | llm | StrOutputParser() ) # 6. 提问 question “如何在LangGraph中实现条件分支” answer rag_chain.invoke(question) print(“问题” question) print(“答案\n” answer)这个管道通过重排序筛选出最相关的3个片段并利用强约束的提示词显著提升了RAG回答的准确性和可靠性。4. 构建智能体AgentLangGraph RAG 工具调用现在我们将前面所学组合起来构建一个更强大的智能体它能根据问题类型自动选择是使用内部知识库RAG回答还是使用搜索引擎工具获取最新信息并能将多步结果进行汇总。4.1 智能体架构设计我们将构建一个具有以下能力的智能体问题分类判断问题是关于内部技术文档还是需要外部实时信息。路由执行内部问题 → 调用RAG检索链。外部问题 → 调用搜索工具。复杂问题 → 可能先搜索再结合RAG结果进行综合。答案合成将检索到的信息组织成连贯、友好的答案。4.2 完整代码实现# 文件tech_doc_agent.py import os from typing import TypedDict, List, Literal from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_chroma import Chroma from langchain_core.prompts import ChatPromptTemplate from langchain_community.tools import TavilySearchResults from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough # ------------------ 0. 状态定义 ------------------ class AgentState(TypedDict): “””智能体的全局状态。“”” user_query: str query_type: Literal[“internal” “external” “hybrid”] | None # 问题类型 internal_context: List[str] # RAG检索到的内部文档 external_context: List[str] # 搜索到的外部信息 final_answer: str # ------------------ 1. 初始化组件 ------------------ llm ChatOpenAI(model“gpt-3.5-turbo”) search_tool TavilySearchResults(max_results3) embeddings OpenAIEmbeddings() # 假设已有一个构建好的内部知识库向量库 vectorstore Chroma(persist_directory“./chroma_db” embedding_functionembeddings) retriever vectorstore.as_retriever(search_kwargs{“k”: 4}) # ------------------ 2. 定义各类提示词模板 ------------------ classification_prompt ChatPromptTemplate.from_template(“”” 请将用户问题分类为以下三种类型之一 - **internal**: 问题明确关于我们内部的技术栈、文档或代码如LangGraph RAG 我们的API规范。 - **external**: 问题需要最新的、实时的外部信息或通用知识如今天的天气 某公司最新新闻。 - **hybrid**: 问题可能既需要内部知识背景也需要外部最新信息补充。 只输出一个单词internal external 或 hybrid。 用户问题{question} 分类结果”””) rag_prompt ChatPromptTemplate.from_template(“”” 你是一个内部技术专家。请严格根据以下内部文档片段来回答问题。 内部文档 {internal_context} 问题{question} 如果文档中没有相关信息请明确说明“内部文档未涵盖此点”。请基于文档给出答案”””) synthesis_prompt ChatPromptTemplate.from_template(“”” 你是一个信息整合助手。用户提出了一个问题我们已经收集了以下信息 用户问题{question} 【内部技术资料】 {internal_context} 【外部网络信息】 {external_context} 请综合以上所有信息生成一个全面、准确、结构清晰的最终答案。如果信息间有冲突以内部技术资料为准。如果某些方面缺乏信息请如实说明。 最终答案”””) # ------------------ 3. 定义图节点 ------------------ def classify_query(state: AgentState): “””节点A问题分类。“”” chain classification_prompt | llm | StrOutputParser() query_type chain.invoke({“question”: state[“user_query”]}).strip().lower() # 简单清理结果 if query_type not in [“internal” “external” “hybrid”]: query_type “hybrid” # 默认混合型 return {“query_type”: query_type} def retrieve_internal_info(state: AgentState): “””节点B检索内部知识库RAG。“”” docs retriever.invoke(state[“user_query”]) context “\n---\n”.join([doc.page_content for doc in docs]) return {“internal_context”: context} def search_external_info(state: AgentState): “””节点C搜索外部信息。“”” results search_tool.invoke(state[“user_query”]) # 提取摘要或内容 external_info [f“来源{r[‘title’]}\n内容{r[‘content’]}” for r in results] return {“external_context”: “\n---\n”.join(external_info)} def route_query(state: AgentState): “””路由节点根据分类结果决定下一步流程。“”” q_type state[“query_type”] if q_type “internal”: return “internal_only” elif q_type “external”: return “external_only” else: # hybrid return “hybrid_process” def answer_with_rag(state: AgentState): “””节点D仅使用RAG回答内部问题。“”” chain rag_prompt | llm | StrOutputParser() answer chain.invoke({ “question”: state[“user_query”] “internal_context”: state[“internal_context”] }) return {“final_answer”: answer} def answer_with_search(state: AgentState): “””节点E仅使用搜索回答外部问题。“”” # 这里可以简单使用搜索结果的摘要也可以让LLM总结 summary_chain ( ChatPromptTemplate.from_template(“请根据以下网络搜索结果总结回答这个问题{question}\n\n结果{results}”) | llm | StrOutputParser() ) answer summary_chain.invoke({ “question”: state[“user_query”] “results”: state[“external_context”] }) return {“final_answer”: answer} def synthesize_hybrid_answer(state: AgentState): “””节点F综合内部和外部信息生成答案。“”” chain synthesis_prompt | llm | StrOutputParser() answer chain.invoke({ “question”: state[“user_query”] “internal_context”: state.get(“internal_context” “无内部资料”) “external_context”: state.get(“external_context” “无外部信息”) }) return {“final_answer”: answer} # ------------------ 4. 构建工作流图 ------------------ workflow StateGraph(AgentState) # 添加所有节点 workflow.add_node(“classify” classify_query) workflow.add_node(“retrieve” retrieve_internal_info) workflow.add_node(“search” search_external_info) workflow.add_node(“answer_rag” answer_with_rag) workflow.add_node(“answer_search” answer_with_search) workflow.add_node(“synthesize” synthesize_hybrid_answer) # 设置入口 workflow.set_entry_point(“classify”) # 添加边和条件路由 workflow.add_edge(“classify” “route_decision”) # 条件路由 workflow.add_conditional_edges( “route_decision” route_query { “internal_only”: “retrieve” “external_only”: “search” “hybrid_process”: [“retrieve” “search”] # 并行执行两个分支 } ) # 定义并行分支后的汇聚点 workflow.add_edge(“retrieve” “answer_rag”) workflow.add_edge(“search” “answer_search”) # 设置最终出口 workflow.add_edge(“answer_rag” END) workflow.add_edge(“answer_search” END) # 对于混合流程需要先并行执行retrieve和search再汇聚到synthesize # 这里需要用到LangGraph的“映射”和“归约”概念为简化我们假设混合流程是串行的。 # 更复杂的并行-汇聚逻辑需要定义子图此处为清晰起见我们修改为串行逻辑。 def route_after_parallel(state): # 这是一个简化处理。实际中应等待并行分支完成。 if state[“query_type”] “hybrid”: return “synthesize” else: return END # 重新构建一个简化版的图避免在示例中引入过于复杂的并行控制 # 我们改为hybrid流程也先检索再搜索最后合成。 workflow_simple StateGraph(AgentState) workflow_simple.add_node(“classify” classify_query) workflow_simple.add_node(“retrieve” retrieve_internal_info) workflow_simple.add_node(“search” search_external_info) workflow_simple.add_node(“answer_rag” answer_with_rag) workflow_simple.add_node(“answer_search” answer_with_search) workflow_simple.add_node(“synthesize” synthesize_hybrid_answer) workflow_simple.set_entry_point(“classify”) workflow_simple.add_edge(“classify” “route_decision_simple”) def route_query_simple(state): q_type state[“query_type”] if q_type “internal”: return “retrieve” elif q_type “external”: return “search” else: # hybrid return “retrieve_for_hybrid” workflow_simple.add_conditional_edges( “route_decision_simple” route_query_simple { “retrieve”: “retrieve” “search”: “search” “retrieve_for_hybrid”: “retrieve” } ) workflow_simple.add_edge(“retrieve” “after_retrieve”) def route_after_retrieve(state): if state[“query_type”] “internal”: return “answer_rag” else: # hybrid return “search_for_hybrid” # hybrid流程继续搜索 workflow_simple.add_conditional_edges(“after_retrieve” route_after_retrieve {“answer_rag”: “answer_rag” “search_for_hybrid”: “search”}) workflow_simple.add_edge(“search” “after_search”) def route_after_search(state): if state[“query_type”] “external”: return “answer_search” else: # hybrid return “synthesize” workflow_simple.add_conditional_edges(“after_search” route_after_search {“answer_search”: “answer_search” “synthesize”: “synthesize”}) workflow_simple.add_edge(“answer_rag” END) workflow_simple.add_edge(“answer_search” END) workflow_simple.add_edge(“synthesize” END) # 编译图 app workflow_simple.compile() # ------------------ 5. 运行智能体 ------------------ if __name__ “__main__”: test_queries [ “LangGraph中的状态State如何定义” # 内部问题 “今天OpenAI有什么最新动态” # 外部问题 “对比一下LangGraph和AutoGen的最新版本特性。” # 混合问题 ] for query in test_queries: print(f“\n{‘’*50}”) print(f“用户问题{query}”) initial_state: AgentState {“user_query”: query} result app.invoke(initial_state) print(f“最终答案\n{result[‘final_answer’]}”) print(f“{‘’*50}”)这个智能体展示了如何将分类、路由、工具调用搜索、RAG检索、信息合成等多个步骤通过LangGraph组织成一个清晰、可控的工作流。你可以通过LangGraph Studio可视化这个工作流直观地调试每个节点的输入输出。5. 环境准备、运行与问题排查5.1 环境准备确保你已安装Python 3.9并创建虚拟环境。# 1. 创建并激活虚拟环境 python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 2. 安装核心库 pip install langgraph langchain langchain-openai langchain-chroma langchain-community pip install openai tiktoken chromadb # 3. 安装可选组件用于搜索和重排序 pip install tavily-python # 搜索工具需要API Key # pip install FlagEmbedding # 如需本地重排序模型5.2 设置API密钥在代码运行前需要设置必要的环境变量。# 在终端中设置或写在 .env 文件中 export OPENAI_API_KEY“sk-...” # 你的OpenAI API Key export TAVILY_API_KEY“tvly-...” # 你的Tavily API Key用于搜索或者在Python代码中设置import os os.environ[“OPENAI_API_KEY”] “sk-...”5.3 运行示例代码将前面的代码示例保存为独立的.py文件。确保你有technical_manual.txt文档用于构建RAG知识库或修改代码加载你自己的文档。按顺序运行# 首先构建向量数据库运行一次即可 python advanced_rag_pipeline.py # 这会创建./chroma_db目录 # 然后运行智能体 python tech_doc_agent.py5.4 常见问题排查表问题现象可能原因排查步骤ModuleNotFoundError依赖未正确安装。1. 确认虚拟环境已激活。2. 运行 pip listAuthenticationError/Invalid API KeyAPI密钥错误或未设置。1. 检查os.environ[‘OPENAI_API_KEY’]是否已正确赋值。2. 确认密钥有效且有余额。3. 对于Tavily等工具检查是否已注册并获取API Key。向量数据库连接失败persist_directory路径错误或权限不足。1. 确认./chroma_db目录存在或可创建。2. 尝试使用绝对路径。RAG检索结果不相关文档分块策略不佳或Embedding模型不合适。1. 调整chunk_size和chunk_overlap。2. 尝试不同的文本分割器如按标题分割。3. 考虑更换或微调Embedding模型。LangGraph图编译或运行错误状态State定义与节点返回值不匹配。1. 检查AgentState的字段类型是否与每个节点返回的字典键完全匹配。2. 使用app.get_graph().draw_mermaid()输出图结构检查逻辑。3. 在节点函数内添加print(state)调试状态变化。Agent回答质量差提示词设计不佳或LLM模型能力不足。1. 优化系统指令和任务描述使其更清晰、更具约束力。2. 增加少样本示例Few-shot Examples。3. 升级到更强大的LLM如GPT-4。4. 引入评估环节量化优化效果。6. 最佳实践与工程建议提示词版本管理像管理代码一样管理你的提示词。使用配置文件如YAML、JSON或专门的提示词管理平台来存储不同版本的Prompt便于A/B测试和回滚。图的可观测性利用LangGraph的内置追踪和LangSmith等工具记录每个节点的输入输出、耗时和异常。这对于调试复杂工作流至关重要。优雅降级与超时控制在调用外部API如LLM、搜索时务必设置超时和重试机制并为失败情况设计备用路径如返回缓存结果、使用更简单的本地模型。测试驱动开发为你的Agent工作流编写单元测试和集成测试。测试应包括正常用例、边界用例、错误处理如网络超时、API限流。安全性输入净化对用户输入进行过滤防止Prompt注入攻击。输出审查对AI生成的内容进行安全检查避免输出有害或不适当信息。权限控制确保工具如文件读写、数据库访问在最小必要权限下运行。成本控制监控LLM的Token消耗。对于非关键路径考虑使用更经济的模型。对RAG检索结果进行摘要或过滤减少送入LLM的上下文长度。从“咒语式”的Prompt尝试到基于状态和图的可控Agent工作流再到工程化的RAG管道构建可靠AI应用的本质是将不确定性封装在可测试、可观测、可迭代的模块中。吴恩达课程的精髓正在于此它提供了一套方法论而非零散的技巧。本文通过一个从简到繁的示例项目串联了Prompt Engineering、LangGraph、RAG和Agent的核心概念与实践。你可以以此为起点深入LangGraph探索子图、检查点持久化状态、人工审批节点等高级特性构建更稳定、支持长对话的Agent。优化RAG尝试不同的Embedding模型、重排序器、查询转换策略甚至用LLM来评估检索质量持续提升知识库的召回率和准确率。设计多Agent系统让多个具备不同技能的Agent协作用LangGraph来编排它们之间的通信和任务分配。真正的挑战不在于理解单个工具而在于如何将它们有机组合解决真实的业务问题。现在你可以尝试用这套框架去改造你手头那个“if-else地狱”的原型或者开始构建一个全新的、真正智能的AI应用了。
)
