1. 项目概述当四个AI角色围坐一张圆桌真正开始“思考”而非“生成”你有没有试过让一个大模型直接写一篇有深度的行业分析报告我试过——结果往往是信息堆砌、逻辑松散、缺乏立场交锋像把维基百科词条和几篇新闻摘要揉在一起再加点漂亮话。这不是模型能力不行而是它天生没有“角色分工”和“观点碰撞”的机制。而这个标题里提到的“Multi-Agent Discussion Panel”正是为了解决这个问题它不靠单个模型硬扛复杂任务而是设计一套协作规则让研究员Researcher专注查资料、专家Expert负责专业解读、批评家Critic专门挑刺质疑、主持人Moderator把控节奏与结论输出——四者形成闭环模拟人类专家小组的真实研讨过程。这背后不是炫技而是对LLM应用边界的实质性突破。LangChain 提供了工具调用、记忆管理、链式编排的基础能力LangGraph 则补上了最关键的一环状态驱动的、可中断可恢复的、带条件分支的多智能体工作流图谱。它让“谁在什么时候做什么、依据什么判断下一步、出错时如何回退”这些原本需要硬编码的逻辑变成可声明、可调试、可复用的图节点。关键词“LLM AI Agent Applications”“LangChain”“LangGraph”“Multi-Agent”“Researcher/Expert/Critic/Moderator”全部指向一个核心如何让AI系统具备结构化协作能力而非单点爆发力。适合正在从单Agent Demo迈向真实业务场景的开发者、技术负责人以及想深入理解AI系统工程化落地路径的架构师。它不教你怎么调API而是带你亲手搭一座桥——连接提示词工程与分布式系统思维的那座桥。2. 整体设计思路为什么必须是四个角色为什么非得用LangGraph2.1 角色划分不是拍脑袋而是对人类认知流程的逆向工程很多人第一反应是“四个Agent是不是太重了两个够不够”我踩过这个坑。早期用ResearcherExpert双角色跑科研选题分析结果Expert总在Researcher还没查完关键论文时就急着下结论输出一堆“可能”“或许”“有待验证”的模糊判断。问题出在缺乏制衡机制。人类专家小组开会没人敢拍板定论除非批评家已经反复质询、主持人确认所有疑点都已澄清。所以四个角色的设计本质是对知识生产流程的解耦Researcher只做一件事——精准检索。它不解释、不评价只返回带来源链接的原始信息片段。它的成功指标不是“回答多好”而是“是否命中权威文献的精确段落”。我实测下来给它配一个带时间过滤的arXiv API 一个能解析PDF表格的本地解析器比让它自己“总结网页”靠谱十倍。Expert严格限定输入源。它只能读Researcher返回的、经主持人筛选过的3-5条高相关性片段。它的提示词里明确写着“禁止引入外部知识所有推论必须标注所依据的原文编号如[1]”。这样逼它把专业能力聚焦在“解读”而非“编造”。Critic它的存在价值不在“挑错”而在“暴露盲区”。它的提示词核心是“请指出Expert分析中未覆盖的Researcher提供的关键事实或逻辑链条断裂处例如Expert说‘A导致B’但Researcher提供的数据只显示A与B相关”。它不负责修正只负责亮红灯。Moderator这才是真正的“导演”。它不参与内容生产只做三件事1接收用户初始问题拆解成Researcher可执行的搜索指令2根据Critic的反馈决定是让Expert重写、让Researcher补充检索还是终止流程3整合最终结论确保每句结论都有对应来源标注。它的提示词里有一行加粗警告“你无权添加任何未被其他Agent输出支持的观点”。提示角色数量不是越多越好而是刚好卡在“能形成有效制衡”的最小集合。少一个制衡失效多一个协调成本指数级上升。我们做过AB测试三角色去掉Critic的结论可信度下降42%基于人工盲评五角色加个Synthesizer的响应延迟增加3.8倍且新增角色90%的输出被Moderator直接丢弃。2.2 LangGraph不是LangChain的升级版而是解决其根本缺陷的“手术刀”LangChain很强大但它本质是线性流水线Chain。你定义好Researcher → Expert → Critic → Moderator的顺序它就一条道走到黑。但真实讨论中Moderator看到Critic指出Expert漏掉了Researcher提供的关键数据[3]必须立刻让Researcher重新聚焦检索[3]相关细节再喂给Expert——这需要状态回溯、条件跳转、循环重入。LangChain的Chain做不到它没有“状态图”的概念。LangGraph的核心价值就是把工作流从“面条代码”变成“状态机”。每个Agent是一个节点Node节点之间不是固定箭头而是由条件函数Conditional Edge决定流向。比如Moderator节点的出口逻辑是def route_after_critic(state): if state[critic_feedback] no_issues: return final_output elif missing_data in state[critic_feedback]: return researcher # 回到Researcher节点重试 else: return expert # 让Expert针对性修改这个函数的返回值直接决定了下一个激活的节点。整个图谱Graph的状态state是共享的所有节点读写同一份字典里面存着当前所有Agent的输出、用户原始问题、中间缓存等。这意味着可调试性你可以随时暂停图谱在任意节点检查state里存了什么就像调试程序时打断点看变量。可恢复性如果Expert节点因超时失败图谱不会崩溃Moderator可以基于现有stateResearcher的输出Critic的反馈直接生成降级结论。可组合性这个四角色Panel本身就是一个节点可以嵌入更大的工作流比如先用Panel做技术可行性分析再把结论喂给另一个“商业决策Panel”。注意LangGraph不是万能胶。它要求你放弃“一步到位”的幻想接受“分步验证、渐进收敛”的新范式。很多团队初期抵触觉得“写个条件函数比写Chain麻烦多了”但当他们第一次用graph.get_state()查出Critic误判是因为Researcher返回的PDF解析乱码从而快速定位到解析器bug时就彻底信服了——LangGraph把隐性错误显性化了。3. 核心细节解析四个Agent的提示词设计、状态结构与工具绑定3.1 提示词不是越长越好而是要“锁死行为边界”新手常犯的错误是给每个Agent塞满“你应该…”“请务必…”的道德说教。结果模型要么无视要么过度迎合导致输出僵硬。我的经验是用结构化输出格式强约束条件比一百句温柔提醒更有效。以下是各角色提示词的核心骨架已脱敏保留关键设计逻辑Researcher提示词关键段落“你是一个学术检索专家。仅执行以下操作解析用户问题提取3个核心学术关键词如‘量子退火’‘组合优化’‘物流调度’调用arXiv_search工具按关键词近3年时间过滤检索对返回的每篇论文调用pdf_parser工具提取摘要、方法论、结论三部分输出严格按JSON格式{ search_keywords: [keyword1, keyword2], sources: [ { title: 论文标题, arxiv_id: 2305.12345, summary: 摘要文本≤200字, method: 方法论描述≤150字, conclusion: 结论≤100字, url: https://arxiv.org/abs/2305.12345 } ] }禁止行为解释术语、比较不同论文、预测未来趋势。你的输出只包含工具返回的原始信息。”Expert提示词关键段落“你是一位[领域]资深研究员如‘计算生物学’。仅基于Researcher提供的sources数组中的内容进行分析。每条分析必须引用具体source索引如‘根据[2]的方法论该模型在…’若sources中无某问题答案必须明确写‘sources未提供相关信息’输出格式## 核心发现 - 发现1引用[1] - 发现2引用[3] ## 待验证假设 - 假设1需验证sources中[2]提及但未证实禁止行为使用‘我认为’‘可能’‘大概’等模糊表述引入任何sources外的知识。”Critic提示词关键段落“你是一个严格的同行评议人。逐条检查Expert输出是否所有‘发现’均有对应source支持若无列出缺失的source索引是否所有‘待验证假设’在sources中确有提及若无标记‘假设虚构’Expert是否将相关性误判为因果性检查‘导致’‘引发’等动词是否有source依据输出格式{ issues: [ {type: missing_source, claim: 发现1, required_source: 1}, {type: causal_error, claim: A导致B, evidence_in_source: A与B相关} ], overall_assessment: critical // 或 acceptable }禁止行为提出改进建议、自行补充信息、评价Expert水平。”Moderator提示词关键段落“你是讨论主持人。执行接收用户问题生成Researcher可用的搜索关键词调用keyword_extractor工具接收Researcher输出选择3个最相关sources喂给Expert接收Critic反馈若issues为空整合Expert输出为终稿若含missing_source将缺失索引传给Researcher重检终稿必须包含用户问题、核心结论加粗、每条结论的source引用如[1]、未解决的待验证项。终极禁令你输出的任何结论必须能在Expert或Researcher的原始输出中找到逐字匹配的依据。”实操心得提示词里“禁止行为”比“应该行为”更重要。我们曾因没写明Expert“禁止使用模糊词”导致它输出“该方法可能提升效率约20%”而Researcher原文只写“在小规模测试中效率提升18%-22%”。Critic根本无法识别这种程度的篡改。加上禁令后错误率归零。3.2 状态State结构所有Agent共享的“会议纪要本”LangGraph的state不是黑盒它是你控制整个Panel的中枢神经。我们定义的state结构经过7轮迭代最终稳定为class PanelState(TypedDict): user_question: str # 用户原始问题不可变 keywords: List[str] # Researcher提取的关键词可被Moderator修正 sources: List[Dict] # Researcher返回的完整source列表只增不删 selected_sources: List[int] # Moderator选定的source索引如[0,2,4] expert_analysis: str # Expert的原始输出Markdown格式 critic_feedback: Dict # Critic的JSON反馈 discussion_history: List[str] # 每次节点执行的日志如Expert completed at 14:22 max_retries: int 2 # 全局重试上限防死循环 current_retry: int 0 # 当前重试次数这个结构的设计哲学是只存必要信息拒绝冗余字段。比如不存“Researcher的中间搜索结果”因为一旦失败重试时会重新生成不存“Expert的思考过程”因为它的输出就是最终分析。所有字段都服务于一个目标让Moderator能基于state做出确定性决策。selected_sources是关键桥梁它把Researcher的海量输出压缩成Expert可消化的3-5条。Moderator的提示词里明确要求“从sources中选择与user_question语义相似度0.85的条目优先选method/conclusion字段信息量大的”。我们用Sentence-BERT做相似度计算阈值0.85是实测平衡覆盖率与噪声的最优值。discussion_history不是日志而是“决策证据链”。当Critic反馈“missing_source for [1]”Moderator查history发现上次选的是[0,2,4]立刻知道要补[1]如果history里有“Researcher failed at retry 2”Moderator就触发降级流程。注意state字段名必须全小写下划线这是LangGraph的硬性要求。曾有团队因用selectedSources驼峰导致图谱静默失败debug三天才发现是命名规范问题。3.3 工具绑定让Agent“手脚并用”而非“空谈理论”四个Agent的能力70%取决于它们能调用什么工具。我们坚持一个原则每个Agent只绑定1-2个高度特化的工具拒绝“全能工具人”。以下是生产环境配置Agent工具名称工具作用关键参数与限制ResearcherarXiv_search调用arXiv官方API支持关键词时间范围分类过滤max_results10强制sort_bysubmittedDatepdf_parser本地部署的PyMuPDF解析器专精于学术PDF公式、表格、参考文献超时15秒失败则跳过该PDF不阻塞流程Expertnone专家不调用工具纯靠LLM分析依赖Researcher提供的结构化数据Criticnone批评家不调用工具纯靠规则校验依赖Expert输出与Researcher source的严格比对Moderatorkeyword_extractor调用微调的BERT模型从user_question中提取学术关键词输出top-3关键词置信度阈值0.6特别说明pdf_parser我们放弃通用网页爬虫因为学术PDF的结构太特殊。比如一篇论文的“方法论”可能在Section 3.2也可能在Appendix B。我们的解析器强制要求识别所有h2标签作为章节标题将“Methodology”“Approach”“Algorithm”等同义词映射到统一字段对含公式的段落保留LaTeX源码如Emc^2不渲染为图片。这样Expert拿到的不是“一段文字”而是带语义标签的结构化数据块分析准确率提升55%。实操心得工具超时设置比功能更重要。arXiv_search设10秒超时pdf_parser设15秒keyword_extractor设3秒。所有工具调用都包在try-except里失败时返回{error: timeout}让Moderator能优雅降级如用关键词直搜Google Scholar替代arXiv。我们统计过网络抖动导致的工具失败占总错误的68%而合理的超时策略让Panel成功率从73%升至94%。4. 实操过程从零搭建Panel含完整代码与调试技巧4.1 环境准备与依赖安装避开版本地狱的三个关键点别跳过这一步。LangGraph 0.1.x和0.2.x的API差异巨大而LangChain 0.1.x与0.2.x的模块路径也完全不同。我们锁定的生产环境组合是# 必须严格匹配的版本2024年Q2实测稳定 pip install langchain0.2.11 pip install langgraph0.2.42 pip install langchain-community0.2.10 # 提供arXiv等社区工具 pip install sentence-transformers2.7.0 # 用于相似度计算 pip install pypdf3.17.2 # PDF解析基础库三个关键避坑点LangChain的llms模块已废弃旧教程里的from langchain.llms import OpenAI会报错。新写法是from langchain_openai import ChatOpenAI # 注意是langchain_openai llm ChatOpenAI(modelgpt-4-turbo, temperature0.3)LangGraph的StateGraph必须指定configurable否则无法在运行时传入API Key等敏感信息。正确初始化from langgraph.graph import StateGraph from langgraph.checkpoint.memory import MemorySaver workflow StateGraph(PanelState) # ... 添加节点和边 app workflow.compile(checkpointerMemorySaver()) # 必须加checkpointer工具必须用tool装饰器注册LangGraph不认LangChain的旧式工具定义。正确写法from langchain_core.tools import tool tool def arXiv_search(query: str) - str: Search arXiv for academic papers. Input: search query string. # 实现代码... return json.dumps(results)提示用pip list | grep lang确认版本用python -c import langgraph; print(langgraph.__version__)验证安装。曾有团队因langgraph装成langgraph-sdk完全不同的包导致StateGraph找不到浪费12小时。4.2 四个Agent节点的实现代码即文档每个Agent节点都是一个纯函数输入state输出dict更新后的state字段。以下是核心实现已简化保留主干逻辑Researcher节点from langchain_core.tools import tool from langchain_community.tools import ArxivQueryRun # 自定义arXiv工具比官方ArxivQueryRun更可控 tool def arxiv_search_tool(keywords: List[str]) - List[Dict]: Search arXiv with strict filters. client arxiv.Client() search arxiv.Search( queryfall:{ AND .join(keywords)}, max_results10, sort_byarxiv.SortCriterion.SubmittedDate, sort_orderarxiv.SortOrder.Descending ) results [] for r in client.results(search): # 解析PDF并提取关键字段 pdf_text extract_pdf_sections(r.pdf_url) # 自定义解析函数 results.append({ title: r.title, arxiv_id: r.entry_id.split(/)[-1], summary: r.summary[:200], method: extract_section(pdf_text, method), # 专用提取函数 conclusion: extract_section(pdf_text, conclusion), url: r.entry_id }) return results def researcher_node(state: PanelState) - Dict: # 调用工具 sources arxiv_search_tool.invoke({keywords: state[keywords]}) # 更新state return { sources: sources, discussion_history: state[discussion_history] [Researcher completed] }Expert节点最简实现凸显提示词威力from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser # 提示词模板已预编译 expert_prompt ChatPromptTemplate.from_messages([ (system, 你是一位[领域]专家...此处为3.1节的Expert提示词), (human, Researcher提供的sources{sources} \n\n请严格按JSON格式输出分析。) ]) expert_chain expert_prompt | llm | StrOutputParser() def expert_node(state: PanelState) - Dict: # 构建sources字符串只取selected_sources selected [state[sources][i] for i in state[selected_sources]] sources_str \n\n.join([f[{i}] {s[title]}\n{s[method]}\n{s[conclusion]} for i, s in enumerate(selected)]) analysis expert_chain.invoke({sources: sources_str}) return { expert_analysis: analysis, discussion_history: state[discussion_history] [Expert completed] }Critic节点规则校验的典范import json import re def critic_node(state: PanelState) - Dict: expert_out state[expert_analysis] sources state[sources] # 提取Expert引用的source索引如[1], [3] cited_indices set() for match in re.finditer(r\[(\d)\], expert_out): idx int(match.group(1)) if idx len(sources): # 防越界 cited_indices.add(idx) # 检查每个cited index是否在sources中真实存在 missing [] for idx in cited_indices: if idx len(sources): missing.append({type: invalid_index, index: idx}) # 检查sources中是否有未被引用的关键信息启发式找含however but limitation的句子 uncited_critical [] for i, s in enumerate(sources): if any(word in s[conclusion].lower() for word in [however, but, limitation]): if i not in cited_indices: uncited_critical.append(i) feedback { issues: missing [{type: uncited_critical, index: i} for i in uncited_critical], overall_assessment: critical if missing or uncited_critical else acceptable } return { critic_feedback: feedback, discussion_history: state[discussion_history] [Critic completed] }Moderator节点决策中枢def moderator_node(state: PanelState) - Dict: # 1. 如果是首次进入生成keywords if not state[keywords]: keywords keyword_extractor.invoke({question: state[user_question]}) return {keywords: keywords} # 2. 如果Critic反馈无问题生成终稿 if state[critic_feedback].get(overall_assessment) acceptable: # 整合Expert输出添加source引用 final_output f**用户问题** {state[user_question]}\n\n final_output f**核心结论**\n{state[expert_analysis]}\n\n final_output **依据来源**\n \n.join([ f- [{i}] {state[sources][i][title]} ({state[sources][i][url]}) for i in state[selected_sources] ]) return {final_output: final_output} # 3. 如果有missing_source重试Researcher if state[current_retry] state[max_retries]: missing_indices [ issue[index] for issue in state[critic_feedback][issues] if issue[type] uncited_critical ] # 补充检索这些source的细节 new_keywords [state[sources][i][title] for i in missing_indices] return { keywords: new_keywords, current_retry: state[current_retry] 1, discussion_history: state[discussion_history] [Moderator triggered retry] } # 4. 达到重试上限降级输出 return { final_output: f**用户问题** {state[user_question]}\n\n f**降级结论** 基于当前可用信息核心结论如下\n{state[expert_analysis]}\n\n f**注意** Critic指出存在未覆盖的关键信息详见feedback建议人工核查。, discussion_history: state[discussion_history] [Moderator triggered fallback] }4.3 图谱编译与执行让四个节点真正“活”起来把节点连成图是LangGraph最直观也最容易出错的环节。完整编译代码from langgraph.graph import StateGraph from langgraph.checkpoint.memory import MemorySaver # 初始化图谱 workflow StateGraph(PanelState) # 添加节点 workflow.add_node(researcher, researcher_node) workflow.add_node(expert, expert_node) workflow.add_node(critic, critic_node) workflow.add_node(moderator, moderator_node) # 设置入口点 workflow.set_entry_point(moderator) # 定义边边节点间的连接 workflow.add_edge(researcher, expert) workflow.add_edge(expert, critic) workflow.add_edge(critic, moderator) # 定义条件边关键 def route_after_moderator(state: PanelState) - str: if keywords not in state or not state[keywords]: # 首次需生成keywords return researcher if final_output in state: # 已完成 return END if state[critic_feedback].get(overall_assessment) acceptable: return END return researcher # 否则重试 workflow.add_conditional_edges( moderator, route_after_moderator, { researcher: researcher, END: END } ) # 编译必须加checkpointer app workflow.compile(checkpointerMemorySaver()) # 执行示例 initial_state PanelState( user_question量子退火在物流路径优化中的实际应用瓶颈是什么, keywords[], sources[], selected_sources[], expert_analysis, critic_feedback{}, discussion_history[], max_retries2, current_retry0 ) # 流式输出每步日志 for output in app.stream(initial_state): for node_name, node_state in output.items(): if discussion_history in node_state: print(f[{node_name}] {node_state[discussion_history][-1]}) if final_output in node_state: print(\n 最终输出 ) print(node_state[final_output]) break实操心得app.stream()比app.invoke()更适合调试。它让你看到每个节点的执行顺序和耗时比如发现pdf_parser在某个PDF上卡了20秒立刻知道要优化解析器。我们还加了app.get_state(config)来实时检查state这比print调试高效十倍。5. 常见问题与排查技巧实录那些文档里不会写的血泪教训5.1 “Critic总说Expert没引用但明明写了[1]”——字符编码陷阱现象Expert输出里有根据[1]的方法论Critic却报missing_source for [1]。根因Researcher返回的sources列表索引是0-based第一个source索引为0而Expert在Markdown里写[1]是1-based。Critic按字符串匹配[1]但sources只有[0]。解决方案在Critic节点里统一转换# Critic中提取引用时 cited_indices set() for match in re.finditer(r\[(\d)\], expert_out): idx int(match.group(1)) - 1 # 强制转为0-based if 0 idx len(sources): cited_indices.add(idx)这个Bug我们花了9小时才定位。教训所有跨Agent的数据索引必须在state层面统一约定我们最终约定所有索引0-based并在每个节点入口做校验。5.2 “Panel跑着跑着就卡死CPU 100%”——隐式循环陷阱现象Moderator触发重试后Researcher返回新sourcesExpert分析Critic又报同样问题无限循环。根因Critic的uncited_critical检测逻辑太宽泛把所有含“however”的句子都当关键信息而Researcher新返回的PDF里恰好也有。解决方案加权重过滤。只标记conclusion字段含“however”且该PDF被arXiv标记为“reviewed”的source# Critic中 if however in s[conclusion].lower() and s.get(is_reviewed, False): uncited_critical.append(i)同时在Moderator重试逻辑里加去重# Moderator中 new_keywords list(set(new_keywords)) # 去重 if len(new_keywords) 3: # 限制每次最多补3个词 new_keywords new_keywords[:3]5.3 “Expert输出格式乱JSON解析失败”——LLM的“创造性”反噬现象Expert节点的StrOutputParser抛出JSONDecodeError因为LLM在JSON末尾加了// end of response注释。解决方案不用StrOutputParser改用JsonOutputParser并加容错from langchain_core.output_parsers import JsonOutputParser # 用JsonOutputParser它会自动清理非JSON字符 parser JsonOutputParser(pydantic_objectExpertOutputSchema) expert_chain expert_prompt | llm | parser # 在节点里加fallback def expert_node(state: PanelState) - Dict: try: analysis expert_chain.invoke({sources: sources_str}) except Exception as e: # 降级用正则提取关键信息 analysis {core_findings: re.findall(r- (.*?)(?\n-|\Z), expert_out)} return {expert_analysis: json.dumps(analysis)}5.4 “响应太慢30秒才出结果”——工具调用的并行化改造现象Researcher串行解析10篇PDF单篇平均5秒总耗时50秒。解决方案用asyncio.gather并发调用pdf_parserimport asyncio async def parse_all_pdfs(pdf_urls: List[str]) - List[Dict]: tasks [pdf_parser_tool.ainvoke({url: url}) for url in pdf_urls] return await asyncio.gather(*tasks, return_exceptionsTrue) # 在Researcher节点中 async def researcher_node_async(state: PanelState) - Dict: # ... 搜索逻辑 sources await parse_all_pdfs([r.pdf_url for r in arxiv_results]) return {sources: sources}效果10篇PDF解析从50秒降至8秒受限于PDF服务器带宽。注意LangGraph原生不支持async节点需用asynchronous装饰器或自定义Executor。5.5 “Moderator总是选错sourcesExpert分析跑偏”——相似度计算的领域适配现象用户问“CRISPR脱靶效应检测方法”Moderator选了篇讲“CRISPR递送载体”的论文因为标题相似度高。解决方案不用标题相似度改用结论段语义相似度。我们训练了一个轻量级BERT微调模型专门对比user_question与source[conclusion]# 加载微调模型 from sentence_transformers import SentenceTransformer model SentenceTransformer(path/to/crispr-conclusion-bert) def select_sources(user_q: str, sources: List[Dict]) - List[int]: # 只计算conclusion相似度 conclusions [s[conclusion] for s in sources] q_emb model.encode([user_q]) c_embs model.encode(conclusions) scores util.cos_sim(q_emb, c_embs)[0].tolist() # 选top-3且score 0.7 return [i for i, s in enumerate(scores) if s 0.7][:3]效果相关性误选率从31%降至6%。代价是增加200MB模型加载内存但值得。6. 性能压测与生产化部署从Demo到每天处理2000次请求6.1 压测结果单实例极限与瓶颈定位我们在AWS t3.xlarge4vCPU/16GB上对Panel做压力测试结果如下并发数平均延迟P95延迟错误率主要瓶颈112.3s14.1s0%arXiv API限速513.8s18.2s0%PDF解析CPU饱和1018.5s32.7s2.1%arXiv超时增多2035.2s68.4s18.3%PDF解析OOM关键发现arXiv API是软瓶颈官方限速1000次/天我们用time.sleep(1)匀速调用但高并发时仍触发429。解决方案加Redis缓存相同keywords的搜索结果缓存1小时。PDF解析是硬瓶颈PyMuPDF吃CPU20并发时CPU 100%内存涨到14GB。解决方案用concurrent.futures.ProcessPoolExecutor隔离解析进程避免GIL阻塞。**