尧图网站建设 
简介本资源是一套面向AI开发者与大模型应用工程师的RAG与Agent智能体实战教程聚焦LangChain框架下的提示词工程、检索增强生成及智能体编排能力培养解决从理论到落地项目开发的关键断层问题。压缩包共235个文件含121个Python核心脚本实现RAG链路、Agent调度、工具调用等、25个中文提示词模板适配Cursor/VSCode Agent等主流AI编程工具、6个PDF技术文档与3个PPTX教学课件另有bin二进制索引文件与yml配置文件支撑本地知识库构建整体22.96MB结构清晰、开箱即用。已有375人学习下载资源提供完整项目代码、可运行的RAGAgent联合案例、中文语境优化的提示词规则集及定期更新的AI编程实践指南助力开发者快速构建具备知识检索、任务分解与自主执行能力的智能应用系统。1. 黑马程序员这套 RAG Agent 实战包不是“讲概念”的课而是能直接跑通一个带知识库问答任务调度的 AI 工作流从 LangChain v0.1.x 到 v0.2.x 的真实迁移踩坑现场你花两小时看完了 LangChain 官方 QuickStart写了个LLMChain能吐出 hello world但当你真想把公司内部的 PDF 手册喂进去、让 AI 根据文档内容回答“报销流程第三步要盖哪个章”再顺手调个钉钉 API 发个审批单——这时候你会发现官方文档里没有load_pdf_and_make_it_work_in_production这个函数。黑马这套实战资源就是为这个“下地干活”时刻准备的它不讲 Transformer 架构不画 attention 矩阵图而是用一个完整闭环项目含可运行源码、清洗好的测试文档集、预置的 FastAPI 接口、带 fallback 的 Agent 调度逻辑把 RAG 的 chunking 策略、embedding 模型选型、retriever 重排序、Agent 的 tool calling 错误兜底、以及最关键的——LangChain 从 v0.1.16 升到 v0.2.10 后Runnable体系重构带来的代码重写点全摊开在你面前。适合已经跑通过 HuggingFace pipeline、会写基础 prompt、但卡在“怎么让 AI 真正按业务规则动起来”的中级开发者。它解决的不是“什么是 RAG”而是“为什么我按教程配了 ChromaDB 却总查不到第 7 页的表格数据”。2. 从零启动解压即跑的项目结构与核心模块定位这套资源压缩包解压后共 4 大主目录/rag_coreRAG 知识库构建与检索、/agent_system多工具协同 Agent、/web_apiFastAPI 封装层、/docs_sample实测用的 12 份 PDF/Markdown 文档。所有模块均基于 Python 3.10依赖锁定在requirements.txt中含langchain0.2.10、langchain-community0.2.9、chromadb0.4.24、sentence-transformers2.2.2。注意它不包含大模型权重文件默认使用 OpenAI API可替换为 Ollama Qwen2-7B 或本地 Llama3-8B这点必须提前确认——否则你会在llm ChatOpenAI(modelgpt-3.5-turbo)这行卡住。2.1 RAG 模块不是简单扔进 Chroma而是分三阶段控制召回质量RAG 的核心不在“存”而在“找得准”。本项目把传统单步 embedding → store → retrieve 拆成三个可调节点Preprocessing Layerrag_core/preprocess.py中的PDFLoaderWithPageNumber类会保留原始 PDF 的页码和标题层级通过pdfplumber提取文本坐标避免“合同第 5 条”被切进两个 chunkEmbedding Chunking Strategyrag_core/chunker.py提供两种模式SemanticChunker基于 sentence-transformers 的相似度聚类chunk size 动态和HierarchicalChunker先按标题分节再对每节做固定长度切分默认启用后者——因为实测中技术文档的章节结构比语义连贯性更重要Retriever Pipelinerag_core/retriever.py不是直接vectorstore.as_retriever()而是封装了MultiQueryRetriever生成 3 个变体 queryContextualCompressionRetriever用EmbeddingsFilter剔除低相关 chunkReRanker调用CohereRerank或本地bge-reranker-base最终返回 top_k5 且 score 0.35 的结果。提示docs_sample/下的hr_policy_v2.3.pdf是专为测试设计的“陷阱文档”——第 12 页有个加粗小字“注本流程自2024年7月1日起废止”但第 3 页正文仍写“请提交至 HRBP”。RAG 模块若未启用重排序大概率召回第 3 页而忽略第 12 页这是检验 pipeline 是否健壮的第一道关。2.2 Agent 模块Tool Calling 不是“调 API”而是带状态机的错误熔断agent_system/目录下的agent_executor.py是整套逻辑的中枢。它没用 LangChain 最新的create_tool_calling_agent而是手写了CustomAgentExecutor类原因很实际官方 agent 在 tool call 失败时默认抛异常终止而生产环境需要降级比如钉钉 API 超时就 fallback 到邮件通知。其核心是三层状态管理Tool Registrytools/下每个.py文件定义一个BaseTool子类必须实现validate_input()参数校验、_run_with_timeout()5s 超时控制、fallback()失败时返回的兜底文案Execution OrchestratorCustomAgentExecutor.run()内部维护execution_history字典记录每步 tool 的输入、输出、耗时、是否 fallback供后续 debugFallback Chain当DingTalkApproveTool返回{status: timeout}自动触发EmailNotifyTool并把原始审批请求存入redis://localhost:6379/agent_fallback_queue供后台 worker 重试。这种设计牺牲了部分简洁性但换来的是线上可追踪、可回滚的执行链路——这正是很多教程忽略的“Agent 真实落地成本”。2.3 Web API 层FastAPI 不是胶水而是承载并发与鉴权的边界网关web_api/main.py表面是标准 FastAPI但关键在三处加固Request Validationschemas.py中QueryRequest模型强制要求session_id: str用于 trace、trace_id: Optional[str]用于链路追踪、timeout: int 30全局超时控制拒绝无 session 的请求Async Streaming Support/rag/query和/agent/run均支持text/event-stream前端可用EventSource接收分块响应避免长请求阻塞Rate Limiting通过slowapi实现 per-session 限流默认 5 req/min配置在web_api/middleware.py且限流 key 包含X-Forwarded-For的前两段 IP防代理穿透。这意味着你不用改一行 LangChain 代码就能把 RAG/Agent 接入现有网关体系——这才是企业级部署的真实形态。3. LangChain v0.2.x 迁移从 Chain 到 Runnable 的重构逻辑与参数映射表LangChain 0.2.x 的最大变化是废弃Chain类全面转向Runnable协议。黑马项目恰好覆盖了这一过渡期其rag_core/pipeline.py提供了清晰的迁移对照v0.1.x 旧写法v0.2.x 新写法关键差异说明LLMChain(llm..., prompt...)promptllmPipeline 操作符RetrievalQA.from_chain_type(...)retriever | (lambda docs: {context: format_docs(docs), question: ...}) | prompt | llmretriever输出是List[Document]必须显式转换为 dict 才能进 prompt旧版隐式处理易出错ConversationalRetrievalChain(...)RunnableWithMessageHistoryconfigurable_fields{session_id: ...}session 管理从memory参数移到config且必须传config{configurable: {session_id: abc123}}output_parserStrOutputParser()llm | StrOutputParser()parser 必须作为 Runnable 链尾不能作为 LLM 初始化参数最典型的翻车点在agent_system/agent_executor.py的invoke()方法v0.1.x 中agent.run(input)直接返回字符串v0.2.x 中agent.invoke({input: input}, config...)返回dict且 key 名变为output非result。项目里用property def final_output(self) - str:封装了兼容层但如果你直接 copy 官方示例就会遇到KeyError: result。# ✅ 正确v0.2.x agent.invoke 返回结构 response agent.invoke( {input: 帮我查报销政策最新版本}, config{configurable: {session_id: sess_20240701}} ) # response 是 dict含 output, intermediate_steps, input final_answer response[output] # 注意是 output不是 result # ❌ 错误沿用 v0.1.x key 名 # final_answer response[result] # KeyError!这个改动看似小却导致整个 Agent 的输出解析逻辑全部重写。黑马项目在agent_system/output_handler.py中提供了parse_agent_output()函数专门处理intermediate_steps中的 tool call 记录提取tool_input和tool_output用于审计——这是 v0.1.x 时代几乎没人做的细节。4. 避坑RAG 与 Agent 开发中五个血泪验证过的典型问题4.1 现象RAG 检索结果总是返回无关文档即使 query 明确指向某页原因ChromaDB默认使用hnsw索引但未设置ef_construction100和ef_search50导致高维向量如bge-m3的 1024 维检索精度骤降同时embedding_model在preprocess.py中被重复初始化造成不同 chunk 的 embedding 向量空间不一致。解决在rag_core/vectorstore.py中修改 Chroma 初始化# 原始错误写法未指定 ef 参数 vectorstore Chroma(embedding_functionembedding_model, persist_directory./chroma_db) # ✅ 正确写法显式控制 hnsw 参数 vectorstore Chroma( embedding_functionembedding_model, persist_directory./chroma_db, client_settingsSettings( anonymized_telemetryFalse, is_persistentTrue, # 关键提升 hnsw 精度 chroma_hnsw_ef_construction100, chroma_hnsw_ef_search50 ) )并确保embedding_model全局单例在rag_core/__init__.py中用lru_cache包装。4.2 现象Agent 调用钉钉 API 时偶发ConnectionResetError但日志无报错原因requests库默认连接池大小为 10当并发 10 时复用旧连接而钉钉网关会主动断开空闲 60s 的连接CustomAgentExecutor的_run_with_timeout()使用threading.Timer但未捕获ConnectionResetError导致整个 execution chain 中断。解决在tools/dingtalk_tool.py中重写requests.Session# ✅ 强制复用连接池设置 keep-alive session requests.Session() adapter requests.adapters.HTTPAdapter( pool_connections20, pool_maxsize20, max_retries3, pool_blockTrue ) session.mount(https://, adapter) session.headers.update({User-Agent: AgentExecutor/1.0}) # 并在 _run_with_timeout 中 catch ConnectionResetError try: response session.post(url, jsonpayload, timeout(3.05, 27)) except requests.exceptions.ConnectionError as e: if ConnectionResetError in str(e): return self.fallback() # 触发降级 raise e4.3 现象MultiQueryRetriever生成的 3 个 query 中2 个完全相同原因llm使用ChatOpenAI时temperature0导致 LLM 输出确定性过高无法生成语义差异 queryMultiQueryRetriever的 prompt 模板未强制要求“query 必须互斥”。解决在rag_core/retriever.py中调整# ✅ 修改 prompt 模板加入约束 MULTI_QUERY_TEMPLATE 你是一个专业的信息检索助手。请基于用户原始问题生成 {num_queries} 个**语义不同但都相关**的搜索 query。 原始问题{question} 要求 - 每个 query 必须独立不能是同义词替换 - 至少一个 query 包含具体数值或日期 - 至少一个 query 使用否定词如“不包括”、“除外” 请直接输出 query每行一个不要编号、不要解释 # 并设置 temperature0.3 保证多样性 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.3)4.4 现象FastAPI/agent/run接口在并发 50 QPS 时出现RuntimeError: Event loop is closed原因langchain的AsyncCallbackHandler在 FastAPI 的async def路由中未正确绑定 event loop当请求被 cancel 或超时时loop 被提前关闭。解决在web_api/main.py的 agent 路由中显式创建新 loop# ✅ 为每个请求创建独立 event loop app.post(/agent/run) async def run_agent(request: QueryRequest): loop asyncio.new_event_loop() asyncio.set_event_loop(loop) try: result await agent_executor.arun( inputrequest.input, config{configurable: {session_id: request.session_id}} ) return {output: result} finally: loop.close() # 确保清理4.5 现象bge-reranker-base重排序后top_k5 的结果反而比未重排序时更差原因BGEReranker的score_threshold0.0默认值过低导致大量低分结果被保留且 reranker 输入的query未做标准化如去除标点、小写化与 embedding 模型训练时的预处理不一致。解决在rag_core/retriever.py中# ✅ 重排序前标准化 query def normalize_query(q: str) - str: return re.sub(r[^\w\s], , q).strip().lower() # ✅ 设置合理阈值 reranker BGEReranker( model_nameBAAI/bge-reranker-base, top_n5, score_threshold0.3 # 低于 0.3 的直接过滤 ) # 使用 normalized_query normalize_query(original_query) docs retriever.get_relevant_documents(normalized_query) reranked_docs reranker.compress_documents( documentsdocs, querynormalized_query )5. 进阶技巧用LangGraph替换CustomAgentExecutor实现可中断、可回溯的 Agent 工作流LangChain 0.2.x 后LangGraph成为构建复杂 Agent 的事实标准。黑马项目虽未内置但agent_system/目录预留了langgraph_adapter.py文件——这是为升级留的接口。我把它补全了用 3 个关键步骤把原有CustomAgentExecutor改造成支持中断、重试、人工审核的 stateful graph5.1 定义 State Schema不只是 input/output还要 track human-in-the-loopfrom typing import Annotated, Sequence, TypedDict import operator class AgentState(TypedDict): input: str output: str intermediate_steps: Annotated[Sequence[tuple], operator.add] needs_review: bool # 是否需人工审核 review_result: str # 审核结果approve/reject/modify last_tool: str # 上次调用的 tool 名 retry_count: int # 当前重试次数5.2 构建 Graph用ConditionalEdge实现动态路由from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver def should_continue(state: AgentState) - str: 根据 state 决定下一步tool call / human review / end if state[needs_review]: return review elif state[retry_count] 3 and error in state[output].lower(): return retry else: return END def call_tool(state: AgentState) - AgentState: # 复用原有 CustomAgentExecutor 的 tool 调用逻辑 tool_name extract_tool_name(state[input]) tool get_tool_by_name(tool_name) result tool._run_with_timeout(state[input]) state[intermediate_steps].append((tool_name, result)) state[last_tool] tool_name return state # 构建 graph workflow StateGraph(AgentState) workflow.add_node(call_tool, call_tool) workflow.add_node(review, human_review_node) # 自定义人工审核节点 workflow.add_node(retry, lambda s: {**s, retry_count: s[retry_count] 1}) workflow.set_entry_point(call_tool) workflow.add_conditional_edges( call_tool, should_continue, { review: review, retry: retry, END: END } ) workflow.add_edge(review, call_tool) workflow.add_edge(retry, call_tool) # 启用 checkpoint支持中断恢复 app workflow.compile(checkpointerMemorySaver())5.3 集成 Human Review用 FastAPI 提供审核端点在web_api/main.py中新增# 审核队列存储待审任务 review_queue [] app.post(/agent/review/queue) async def queue_for_review(task: ReviewTask): review_queue.append({ task_id: str(uuid4()), state: task.state, timestamp: datetime.now().isoformat() }) return {status: queued, task_id: review_queue[-1][task_id]} app.get(/agent/review/pending) async def get_pending_reviews(): return {pending: review_queue[:5]} # 只返回前5条 app.post(/agent/review/{task_id}) async def submit_review(task_id: str, result: ReviewResult): # 找到对应 task 并更新 state for item in review_queue: if item[task_id] task_id: item[review_result] result.result # 触发 graph 继续执行 app.invoke(item[state], config{thread_id: task_id}) review_queue.remove(item) break return {status: processed}这样当 Agent 调用支付接口时自动进入review节点前端弹出审核弹窗审核员点击“批准”graph 自动 resume若点击“拒绝”则触发retry节点Agent 用备用渠道如邮件重试。整个过程 state 可存可查MemorySaver()保证服务重启后不丢上下文。从那以后我每次设计 Agent都强制走一遍LangGraph的 state schema 定义——哪怕初期只用单节点也要把needs_review、retry_count、last_tool这些字段写进 TypedDict。因为真正的业务复杂度从来不是“能调几个 API”而是“什么时候该停、谁来拍板、失败了往哪退”。希望帮到你。本文还有配套的精品资源点击获取 LangChain 0.2.x RAG+Agent 实战：从知识库问答到生产级任务调度
简介本资源是一套面向AI开发者与大模型应用工程师的RAG与Agent智能体实战教程聚焦LangChain框架下的提示词工程、检索增强生成及智能体编排能力培养解决从理论到落地项目开发的关键断层问题。压缩包共235个文件含121个Python核心脚本实现RAG链路、Agent调度、工具调用等、25个中文提示词模板适配Cursor/VSCode Agent等主流AI编程工具、6个PDF技术文档与3个PPTX教学课件另有bin二进制索引文件与yml配置文件支撑本地知识库构建整体22.96MB结构清晰、开箱即用。已有375人学习下载资源提供完整项目代码、可运行的RAGAgent联合案例、中文语境优化的提示词规则集及定期更新的AI编程实践指南助力开发者快速构建具备知识检索、任务分解与自主执行能力的智能应用系统。1. 黑马程序员这套 RAG Agent 实战包不是“讲概念”的课而是能直接跑通一个带知识库问答任务调度的 AI 工作流从 LangChain v0.1.x 到 v0.2.x 的真实迁移踩坑现场你花两小时看完了 LangChain 官方 QuickStart写了个LLMChain能吐出 hello world但当你真想把公司内部的 PDF 手册喂进去、让 AI 根据文档内容回答“报销流程第三步要盖哪个章”再顺手调个钉钉 API 发个审批单——这时候你会发现官方文档里没有load_pdf_and_make_it_work_in_production这个函数。黑马这套实战资源就是为这个“下地干活”时刻准备的它不讲 Transformer 架构不画 attention 矩阵图而是用一个完整闭环项目含可运行源码、清洗好的测试文档集、预置的 FastAPI 接口、带 fallback 的 Agent 调度逻辑把 RAG 的 chunking 策略、embedding 模型选型、retriever 重排序、Agent 的 tool calling 错误兜底、以及最关键的——LangChain 从 v0.1.16 升到 v0.2.10 后Runnable体系重构带来的代码重写点全摊开在你面前。适合已经跑通过 HuggingFace pipeline、会写基础 prompt、但卡在“怎么让 AI 真正按业务规则动起来”的中级开发者。它解决的不是“什么是 RAG”而是“为什么我按教程配了 ChromaDB 却总查不到第 7 页的表格数据”。2. 从零启动解压即跑的项目结构与核心模块定位这套资源压缩包解压后共 4 大主目录/rag_coreRAG 知识库构建与检索、/agent_system多工具协同 Agent、/web_apiFastAPI 封装层、/docs_sample实测用的 12 份 PDF/Markdown 文档。所有模块均基于 Python 3.10依赖锁定在requirements.txt中含langchain0.2.10、langchain-community0.2.9、chromadb0.4.24、sentence-transformers2.2.2。注意它不包含大模型权重文件默认使用 OpenAI API可替换为 Ollama Qwen2-7B 或本地 Llama3-8B这点必须提前确认——否则你会在llm ChatOpenAI(modelgpt-3.5-turbo)这行卡住。2.1 RAG 模块不是简单扔进 Chroma而是分三阶段控制召回质量RAG 的核心不在“存”而在“找得准”。本项目把传统单步 embedding → store → retrieve 拆成三个可调节点Preprocessing Layerrag_core/preprocess.py中的PDFLoaderWithPageNumber类会保留原始 PDF 的页码和标题层级通过pdfplumber提取文本坐标避免“合同第 5 条”被切进两个 chunkEmbedding Chunking Strategyrag_core/chunker.py提供两种模式SemanticChunker基于 sentence-transformers 的相似度聚类chunk size 动态和HierarchicalChunker先按标题分节再对每节做固定长度切分默认启用后者——因为实测中技术文档的章节结构比语义连贯性更重要Retriever Pipelinerag_core/retriever.py不是直接vectorstore.as_retriever()而是封装了MultiQueryRetriever生成 3 个变体 queryContextualCompressionRetriever用EmbeddingsFilter剔除低相关 chunkReRanker调用CohereRerank或本地bge-reranker-base最终返回 top_k5 且 score 0.35 的结果。提示docs_sample/下的hr_policy_v2.3.pdf是专为测试设计的“陷阱文档”——第 12 页有个加粗小字“注本流程自2024年7月1日起废止”但第 3 页正文仍写“请提交至 HRBP”。RAG 模块若未启用重排序大概率召回第 3 页而忽略第 12 页这是检验 pipeline 是否健壮的第一道关。2.2 Agent 模块Tool Calling 不是“调 API”而是带状态机的错误熔断agent_system/目录下的agent_executor.py是整套逻辑的中枢。它没用 LangChain 最新的create_tool_calling_agent而是手写了CustomAgentExecutor类原因很实际官方 agent 在 tool call 失败时默认抛异常终止而生产环境需要降级比如钉钉 API 超时就 fallback 到邮件通知。其核心是三层状态管理Tool Registrytools/下每个.py文件定义一个BaseTool子类必须实现validate_input()参数校验、_run_with_timeout()5s 超时控制、fallback()失败时返回的兜底文案Execution OrchestratorCustomAgentExecutor.run()内部维护execution_history字典记录每步 tool 的输入、输出、耗时、是否 fallback供后续 debugFallback Chain当DingTalkApproveTool返回{status: timeout}自动触发EmailNotifyTool并把原始审批请求存入redis://localhost:6379/agent_fallback_queue供后台 worker 重试。这种设计牺牲了部分简洁性但换来的是线上可追踪、可回滚的执行链路——这正是很多教程忽略的“Agent 真实落地成本”。2.3 Web API 层FastAPI 不是胶水而是承载并发与鉴权的边界网关web_api/main.py表面是标准 FastAPI但关键在三处加固Request Validationschemas.py中QueryRequest模型强制要求session_id: str用于 trace、trace_id: Optional[str]用于链路追踪、timeout: int 30全局超时控制拒绝无 session 的请求Async Streaming Support/rag/query和/agent/run均支持text/event-stream前端可用EventSource接收分块响应避免长请求阻塞Rate Limiting通过slowapi实现 per-session 限流默认 5 req/min配置在web_api/middleware.py且限流 key 包含X-Forwarded-For的前两段 IP防代理穿透。这意味着你不用改一行 LangChain 代码就能把 RAG/Agent 接入现有网关体系——这才是企业级部署的真实形态。3. LangChain v0.2.x 迁移从 Chain 到 Runnable 的重构逻辑与参数映射表LangChain 0.2.x 的最大变化是废弃Chain类全面转向Runnable协议。黑马项目恰好覆盖了这一过渡期其rag_core/pipeline.py提供了清晰的迁移对照v0.1.x 旧写法v0.2.x 新写法关键差异说明LLMChain(llm..., prompt...)promptllmPipeline 操作符RetrievalQA.from_chain_type(...)retriever | (lambda docs: {context: format_docs(docs), question: ...}) | prompt | llmretriever输出是List[Document]必须显式转换为 dict 才能进 prompt旧版隐式处理易出错ConversationalRetrievalChain(...)RunnableWithMessageHistoryconfigurable_fields{session_id: ...}session 管理从memory参数移到config且必须传config{configurable: {session_id: abc123}}output_parserStrOutputParser()llm | StrOutputParser()parser 必须作为 Runnable 链尾不能作为 LLM 初始化参数最典型的翻车点在agent_system/agent_executor.py的invoke()方法v0.1.x 中agent.run(input)直接返回字符串v0.2.x 中agent.invoke({input: input}, config...)返回dict且 key 名变为output非result。项目里用property def final_output(self) - str:封装了兼容层但如果你直接 copy 官方示例就会遇到KeyError: result。# ✅ 正确v0.2.x agent.invoke 返回结构 response agent.invoke( {input: 帮我查报销政策最新版本}, config{configurable: {session_id: sess_20240701}} ) # response 是 dict含 output, intermediate_steps, input final_answer response[output] # 注意是 output不是 result # ❌ 错误沿用 v0.1.x key 名 # final_answer response[result] # KeyError!这个改动看似小却导致整个 Agent 的输出解析逻辑全部重写。黑马项目在agent_system/output_handler.py中提供了parse_agent_output()函数专门处理intermediate_steps中的 tool call 记录提取tool_input和tool_output用于审计——这是 v0.1.x 时代几乎没人做的细节。4. 避坑RAG 与 Agent 开发中五个血泪验证过的典型问题4.1 现象RAG 检索结果总是返回无关文档即使 query 明确指向某页原因ChromaDB默认使用hnsw索引但未设置ef_construction100和ef_search50导致高维向量如bge-m3的 1024 维检索精度骤降同时embedding_model在preprocess.py中被重复初始化造成不同 chunk 的 embedding 向量空间不一致。解决在rag_core/vectorstore.py中修改 Chroma 初始化# 原始错误写法未指定 ef 参数 vectorstore Chroma(embedding_functionembedding_model, persist_directory./chroma_db) # ✅ 正确写法显式控制 hnsw 参数 vectorstore Chroma( embedding_functionembedding_model, persist_directory./chroma_db, client_settingsSettings( anonymized_telemetryFalse, is_persistentTrue, # 关键提升 hnsw 精度 chroma_hnsw_ef_construction100, chroma_hnsw_ef_search50 ) )并确保embedding_model全局单例在rag_core/__init__.py中用lru_cache包装。4.2 现象Agent 调用钉钉 API 时偶发ConnectionResetError但日志无报错原因requests库默认连接池大小为 10当并发 10 时复用旧连接而钉钉网关会主动断开空闲 60s 的连接CustomAgentExecutor的_run_with_timeout()使用threading.Timer但未捕获ConnectionResetError导致整个 execution chain 中断。解决在tools/dingtalk_tool.py中重写requests.Session# ✅ 强制复用连接池设置 keep-alive session requests.Session() adapter requests.adapters.HTTPAdapter( pool_connections20, pool_maxsize20, max_retries3, pool_blockTrue ) session.mount(https://, adapter) session.headers.update({User-Agent: AgentExecutor/1.0}) # 并在 _run_with_timeout 中 catch ConnectionResetError try: response session.post(url, jsonpayload, timeout(3.05, 27)) except requests.exceptions.ConnectionError as e: if ConnectionResetError in str(e): return self.fallback() # 触发降级 raise e4.3 现象MultiQueryRetriever生成的 3 个 query 中2 个完全相同原因llm使用ChatOpenAI时temperature0导致 LLM 输出确定性过高无法生成语义差异 queryMultiQueryRetriever的 prompt 模板未强制要求“query 必须互斥”。解决在rag_core/retriever.py中调整# ✅ 修改 prompt 模板加入约束 MULTI_QUERY_TEMPLATE 你是一个专业的信息检索助手。请基于用户原始问题生成 {num_queries} 个**语义不同但都相关**的搜索 query。 原始问题{question} 要求 - 每个 query 必须独立不能是同义词替换 - 至少一个 query 包含具体数值或日期 - 至少一个 query 使用否定词如“不包括”、“除外” 请直接输出 query每行一个不要编号、不要解释 # 并设置 temperature0.3 保证多样性 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.3)4.4 现象FastAPI/agent/run接口在并发 50 QPS 时出现RuntimeError: Event loop is closed原因langchain的AsyncCallbackHandler在 FastAPI 的async def路由中未正确绑定 event loop当请求被 cancel 或超时时loop 被提前关闭。解决在web_api/main.py的 agent 路由中显式创建新 loop# ✅ 为每个请求创建独立 event loop app.post(/agent/run) async def run_agent(request: QueryRequest): loop asyncio.new_event_loop() asyncio.set_event_loop(loop) try: result await agent_executor.arun( inputrequest.input, config{configurable: {session_id: request.session_id}} ) return {output: result} finally: loop.close() # 确保清理4.5 现象bge-reranker-base重排序后top_k5 的结果反而比未重排序时更差原因BGEReranker的score_threshold0.0默认值过低导致大量低分结果被保留且 reranker 输入的query未做标准化如去除标点、小写化与 embedding 模型训练时的预处理不一致。解决在rag_core/retriever.py中# ✅ 重排序前标准化 query def normalize_query(q: str) - str: return re.sub(r[^\w\s], , q).strip().lower() # ✅ 设置合理阈值 reranker BGEReranker( model_nameBAAI/bge-reranker-base, top_n5, score_threshold0.3 # 低于 0.3 的直接过滤 ) # 使用 normalized_query normalize_query(original_query) docs retriever.get_relevant_documents(normalized_query) reranked_docs reranker.compress_documents( documentsdocs, querynormalized_query )5. 进阶技巧用LangGraph替换CustomAgentExecutor实现可中断、可回溯的 Agent 工作流LangChain 0.2.x 后LangGraph成为构建复杂 Agent 的事实标准。黑马项目虽未内置但agent_system/目录预留了langgraph_adapter.py文件——这是为升级留的接口。我把它补全了用 3 个关键步骤把原有CustomAgentExecutor改造成支持中断、重试、人工审核的 stateful graph5.1 定义 State Schema不只是 input/output还要 track human-in-the-loopfrom typing import Annotated, Sequence, TypedDict import operator class AgentState(TypedDict): input: str output: str intermediate_steps: Annotated[Sequence[tuple], operator.add] needs_review: bool # 是否需人工审核 review_result: str # 审核结果approve/reject/modify last_tool: str # 上次调用的 tool 名 retry_count: int # 当前重试次数5.2 构建 Graph用ConditionalEdge实现动态路由from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver def should_continue(state: AgentState) - str: 根据 state 决定下一步tool call / human review / end if state[needs_review]: return review elif state[retry_count] 3 and error in state[output].lower(): return retry else: return END def call_tool(state: AgentState) - AgentState: # 复用原有 CustomAgentExecutor 的 tool 调用逻辑 tool_name extract_tool_name(state[input]) tool get_tool_by_name(tool_name) result tool._run_with_timeout(state[input]) state[intermediate_steps].append((tool_name, result)) state[last_tool] tool_name return state # 构建 graph workflow StateGraph(AgentState) workflow.add_node(call_tool, call_tool) workflow.add_node(review, human_review_node) # 自定义人工审核节点 workflow.add_node(retry, lambda s: {**s, retry_count: s[retry_count] 1}) workflow.set_entry_point(call_tool) workflow.add_conditional_edges( call_tool, should_continue, { review: review, retry: retry, END: END } ) workflow.add_edge(review, call_tool) workflow.add_edge(retry, call_tool) # 启用 checkpoint支持中断恢复 app workflow.compile(checkpointerMemorySaver())5.3 集成 Human Review用 FastAPI 提供审核端点在web_api/main.py中新增# 审核队列存储待审任务 review_queue [] app.post(/agent/review/queue) async def queue_for_review(task: ReviewTask): review_queue.append({ task_id: str(uuid4()), state: task.state, timestamp: datetime.now().isoformat() }) return {status: queued, task_id: review_queue[-1][task_id]} app.get(/agent/review/pending) async def get_pending_reviews(): return {pending: review_queue[:5]} # 只返回前5条 app.post(/agent/review/{task_id}) async def submit_review(task_id: str, result: ReviewResult): # 找到对应 task 并更新 state for item in review_queue: if item[task_id] task_id: item[review_result] result.result # 触发 graph 继续执行 app.invoke(item[state], config{thread_id: task_id}) review_queue.remove(item) break return {status: processed}这样当 Agent 调用支付接口时自动进入review节点前端弹出审核弹窗审核员点击“批准”graph 自动 resume若点击“拒绝”则触发retry节点Agent 用备用渠道如邮件重试。整个过程 state 可存可查MemorySaver()保证服务重启后不丢上下文。从那以后我每次设计 Agent都强制走一遍LangGraph的 state schema 定义——哪怕初期只用单节点也要把needs_review、retry_count、last_tool这些字段写进 TypedDict。因为真正的业务复杂度从来不是“能调几个 API”而是“什么时候该停、谁来拍板、失败了往哪退”。希望帮到你。本文还有配套的精品资源点击获取