LangChain集成百度千帆SDK：从零构建可落地的LLM应用实战指南

1. 从想法到产品为什么LLM应用落地这么难最近和不少同行聊发现一个挺普遍的现象大家都能用ChatGPT的网页版玩出花来但一旦想把大模型能力集成到自己的业务系统里或者想做一个能稳定对外服务的AI应用立刻就卡住了。不是API调用几次就报错就是效果时好时坏再不然就是整个流程跑起来又慢又笨重。这感觉就像你有一台顶级跑车的发动机大模型但不知道怎么给它装上轮子、方向盘和刹车让它能在真实的道路上安全平稳地跑起来。这个“装轮子”的过程就是我们常说的LLM应用落地。它远不止是调个API那么简单。你至少得面对几个头疼的问题第一是工程化怎么把模型调用、上下文管理、对话历史、工具调用这些琐碎但关键的部分封装成可靠的服务第二是效果稳定性如何确保每次对话的质量处理模型可能出现的“胡言乱语”第三是成本与效率如何管理token消耗、优化响应速度第四是生态对接你的应用可能需要查数据库、调外部API、处理文件这些能力如何与LLM无缝结合直接裸写代码去调用模型API很快你就会陷入无尽的细节泥潭。这时候一个成熟的框架就显得至关重要。它就像一套标准的汽车底盘和传动系统能帮你省去大量重复造轮子的时间让你专注于上层业务逻辑和体验优化。在众多框架中LangChain凭借其设计理念和活跃的生态成为了很多开发者的首选。它提供了一套高层次的抽象将LLM、记忆、工具链、数据检索等概念模块化让构建复杂AI应用的逻辑变得清晰。而框架的威力需要结合强大的模型才能发挥。国内在LLM服务化方面百度智能云千帆平台是一个重要的选择。它提供了包括文心一言系列在内的多种主流模型的一站式服务具备稳定的API、完善的监控和相对友好的计费方式。那么一个很自然的想法就是能否用LangChain这套“好底盘”去接入千帆平台提供的“强引擎”呢答案是肯定的而且这正是快速构建可靠LLM应用的捷径。本文将围绕“通过Langchain接入千帆SDK”这个核心动作拆解从零开始将一个LLM想法快速落地成可运行、可扩展应用的全过程。我会结合真实的项目经验不仅告诉你步骤“是什么”更会重点解释每个环节“为什么”要这么做以及在实际操作中容易踩哪些坑、如何规避。2. 环境奠基构建一个可复现的Python工作空间在动手写一行LLM代码之前搭建一个干净、隔离、可复现的Python环境是至关重要的一步这也是很多新手容易忽略导致后期依赖冲突、环境崩溃的根源。我们追求的不仅仅是“能跑”而是“在任何机器上都能以相同的方式跑起来”。2.1 虚拟环境项目的安全隔离舱强烈建议为每个LLM项目创建独立的虚拟环境。这能确保项目A所需的LangChain版本不会与项目B的版本冲突。这里我推荐使用venv它是Python 3.3自带的标准库工具无需额外安装。打开你的终端Windows用CMD或PowerShellMac/Linux用Terminal进入你的项目目录然后执行# 创建名为 llm_app_env 的虚拟环境 python -m venv llm_app_env创建完成后激活它Windows (CMD):llm_app_env\Scripts\activate.batWindows (PowerShell):llm_app_env\Scripts\Activate.ps1Mac/Linux:source llm_app_env/bin/activate激活后你的命令行提示符前通常会显示环境名(llm_app_env)这表示你已进入该隔离环境。后续所有pip install操作都只会影响这个环境。注意在Windows PowerShell中执行激活脚本可能会因执行策略而报错。如果遇到可以以管理员身份打开PowerShell先执行Set-ExecutionPolicy RemoteSigned -Scope CurrentUser更改策略选择Y或者直接使用CMD。2.2 依赖管理用requirements.txt锁定一切在虚拟环境中我们开始安装核心依赖。首先升级pip到最新版以确保安装过程顺利pip install --upgrade pip接下来一次性安装我们项目所需的包。你可以手动执行下面的pip install命令但我更推荐的做法是先创建一个requirements.txt文件来管理依赖。这是一个好习惯能让你的项目依赖一目了然也方便他人复现。在你的项目根目录下创建requirements.txt文件并填入以下内容langchain0.1.0 langchain-community0.0.10 qianfan0.3.0 python-dotenv1.0.0然后在激活的虚拟环境中运行pip install -r requirements.txt这里解释一下每个包的作用langchain: 核心框架提供了构建链Chains、代理Agents等核心抽象。langchain-community: LangChain社区维护的第三方集成包其中包含了我们要用的千帆QianfanLLM集成。从LangChain 0.1.x版本开始许多第三方模型集成被移到了这个独立的包中以保持核心框架的轻量。qianfan: 百度千帆平台的官方Python SDK。虽然langchain-community中的集成会封装调用逻辑但某些高级功能或直接调用千帆其他服务如模型管理、批量推理时可能需要它。python-dotenv: 用于从.env文件加载环境变量。这是管理API密钥等敏感信息的标准做法切勿将密钥硬编码在代码中安装完成后可以通过pip list命令检查已安装的包及其版本。2.3 密钥配置安全第一切勿泄露接下来是最关键也最敏感的一步——配置千帆平台的访问密钥。你需要登录 百度智能云千帆控制台 在“应用接入”中创建一个应用从而获取API Key和Secret Key。在项目根目录下创建一个名为.env的文件注意开头有个点。这个文件应该被添加到.gitignore中绝对不要提交到版本控制系统。在.env文件中填入你的密钥QIANFAN_AK你的API_Key QIANFAN_SK你的Secret_Key这样你的代码将通过环境变量读取这些密钥既安全又方便在不同环境开发、测试、生产间切换。3. 建立连接从第一行代码到第一次对话环境就绪后让我们编写最简单的代码验证整个链路是否通畅。这个过程能帮你快速建立信心并理解LangChain接入千帆的基本模式。3.1 初始化LLM选择你的模型引擎在项目目录下创建一个Python文件例如first_chat.py。首先加载环境变量并导入必要的模块。import os from dotenv import load_dotenv from langchain_community.llms import QianfanLLMEndpoint # 1. 加载 .env 文件中的环境变量 load_dotenv() # 2. 从环境变量获取密钥确保 .env 文件已正确配置 qianfan_ak os.getenv(QIANFAN_AK) qianfan_sk os.getenv(QIANFAN_SK) if not qianfan_ak or not qianfan_sk: raise ValueError(请在 .env 文件中设置 QIANFAN_AK 和 QIANFAN_SK 环境变量。) # 3. 初始化千帆LLM llm QianfanLLMEndpoint( qianfan_akqianfan_ak, qianfan_skqianfan_sk, modelERNIE-Bot-turbo, # 指定模型例如文心一言Turbo版 endpoint, # 通常留空使用默认端点。如需特定定制化模型可在此填写其专属endpoint。 )这里有几个关键点需要解释QianfanLLMEndpoint: 这是langchain-community中为千帆模型提供的集成类。它封装了与千帆API的通信细节提供了一个标准的LangChain LLM接口。model参数: 这是最重要的参数之一决定了你使用千帆平台上的哪个模型。例如ERNIE-Bot-turbo: 文心一言Turbo版响应速度快性价比高适用于大部分对话场景。ERNIE-Bot: 文心一言标准版能力更均衡。ERNIE-Bot-4: 文心一言4.0版本理解与生成能力更强。Llama-2-7b-chat等: 千帆也提供了开源模型的服务。你需要去千帆控制台的“模型服务”页面查看你已开通或可用的模型名称列表。endpoint参数: 对于平台提供的标准模型通常留空即可SDK会根据model名称自动拼装正确的API地址。只有当你使用了“模型服务”中自己部署的定制模型时才需要填写其提供的专属Endpoint。3.2 发起调用理解同步与异步初始化完成后调用就非常简单了。LangChain的LLM对象最核心的方法是invoke同步和ainvoke异步。# 4. 同步调用示例 print( 同步调用 ) question 请用一句话介绍你自己。 try: response llm.invoke(question) print(f问{question}) print(f答{response}) except Exception as e: print(f调用出错{e}) # 5. 可选异步调用示例 - 适用于高并发Web应用 import asyncio print(\n 异步调用 ) async def async_chat(): async_question 异步调用的感觉怎么样 try: async_response await llm.ainvoke(async_question) print(f问{async_question}) print(f答{async_response}) except Exception as e: print(f异步调用出错{e}) # 运行异步函数 asyncio.run(async_chat())运行这个脚本 (python first_chat.py)如果一切配置正确你将看到模型的回复。这标志着从你的代码到千帆大模型的服务通道已经成功建立。第一个坑与技巧你可能会遇到ImportError: cannot import name QianfanLLMEndpoint from langchain_community.llms。这是因为LangChain版本和集成包结构变化较快。如果遇到此问题可以尝试以下方法检查langchain-community版本尝试升级到最新pip install -U langchain-community。查看官方文档或源码类名可能已更新例如变为QianfanLLM。你可以通过from langchain_community.llms import *然后print([c for c in dir() if ‘Qianfan’ in c])来查看当前可用的类名。作为备选方案你可以直接使用千帆官方SDK (qianfan) 进行调用虽然会失去LangChain的一些便利但更稳定。代码如下import qianfan chat_comp qianfan.ChatCompletion() resp chat_comp.do(messages[{“role”: “user”, “content”: “你好”}], model“ERNIE-Bot-turbo”) print(resp[“body”][“result”])4. 超越单次问答构建可记忆、可检索的对话链一次性的问答用处有限。真实的LLM应用需要记忆上下文、处理复杂逻辑。这就是LangChain核心概念“链”Chain和“记忆”Memory发挥作用的地方。4.1 创建对话链让AI记住之前说了什么让我们构建一个能进行多轮对话的简单聊天机器人。我们需要两样东西一个LLM和一块“记忆内存”。import os from dotenv import load_dotenv from langchain_community.llms import QianfanLLMEndpoint from langchain.chains import ConversationChain from langchain.memory import ConversationBufferMemory load_dotenv() # 初始化LLM (同上) llm QianfanLLMEndpoint( qianfan_akos.getenv(QIANFAN_AK), qianfan_skos.getenv(QIANFAN_SK), modelERNIE-Bot-turbo, ) # 初始化记忆体ConversationBufferMemory 会保存完整的对话历史 memory ConversationBufferMemory() # 创建对话链 conversation_chain ConversationChain( llmllm, memorymemory, verboseTrue, # 设置为True可以看到链的详细执行过程调试时非常有用 ) # 进行多轮对话 print(开始对话输入‘退出’结束) while True: user_input input(\n你) if user_input.lower() 退出: print(对话结束。) break # 调用链的predict方法传入当前输入 response conversation_chain.predict(inputuser_input) print(fAI{response})运行这段代码你会发现AI能够基于之前的对话内容进行回复。例如你先说“我叫小明”再问“我的名字是什么”它应该能回答出来。关键就在于ConversationBufferMemory对象它自动地将每轮对话的输入和输出拼接起来作为下一次请求的“上下文”或“历史消息”传递给LLM。重要提示verboseTrue会在控制台打印出LangChain内部执行的“提示词模板”Prompt Template和传递给模型的完整信息。这是理解LangChain工作原理和调试问题的黄金工具。你会看到类似 Entering new ConversationChain chain...和Prompt after formatting:的输出里面包含了模型实际接收到的文本。通过观察这个你可以清楚地知道记忆是如何被格式化成提示词的。4.2 管理上下文长度避免Token超限的陷阱ConversationBufferMemory虽然简单但有一个致命缺点它会无限制地增长对话历史。大模型API通常有上下文长度限制例如4K、8K、16K tokens。当历史对话超过这个限制API就会报错。解决方案是使用能管理窗口的记忆体例如ConversationBufferWindowMemory。它只保留最近K轮对话。from langchain.memory import ConversationBufferWindowMemory # 只保留最近3轮对话的记忆 window_memory ConversationBufferWindowMemory(k3) conversation_chain_window ConversationChain( llmllm, memorywindow_memory, verboseFalse ) # 测试进行超过3轮的对话观察AI是否“忘记”了最早的内容。另一个更智能的方案是ConversationSummaryMemory。它不会保存原始对话而是让LLM定期对之前的对话内容进行总结只把总结摘要作为历史上下文。这能极大地节省token但会引入额外的模型调用和总结可能失真的风险。from langchain.memory import ConversationSummaryMemory summary_memory ConversationSummaryMemory(llmllm) summary_chain ConversationChain(llmllm, memorysummary_memory, verboseTrue) # 进行多轮对话后查看 memory.buffer 属性里面存储的是总结文本而非原始对话。选择策略对于短对话、调试场景用ConversationBufferMemory。对于需要固定近期记忆的聊天应用用ConversationBufferWindowMemory。对于长文档分析、长程对话且对历史细节要求不高的场景可以尝试ConversationSummaryMemory但要注意测试总结效果。4.3 设计提示词模板引导模型扮演特定角色默认的对话链提示词比较通用。在实际应用中我们通常需要引导模型扮演特定角色、遵循特定格式。这就需要用到提示词模板PromptTemplate。假设我们要创建一个“IT技术支持专家”助手。from langchain.prompts import PromptTemplate from langchain.chains import LLMChain # 1. 定义专属提示词模板 support_template 你是一名专业的IT技术支持专家负责回答用户关于电脑、网络、软件方面的技术问题。 你的回答应该专业、清晰、分步骤并且充满耐心。 如果用户的问题信息不足你应该主动询问关键细节。 如果问题超出你的知识范围应如实告知并建议可能的解决方向。 之前的对话历史 {history} 用户当前问题{input} 技术支持专家 prompt PromptTemplate( input_variables[history, input], # 模板中需要被替换的变量 templatesupport_template ) # 2. 创建带有自定义提示词和记忆的链 from langchain.memory import ConversationBufferWindowMemory memory ConversationBufferWindowMemory(k5, memory_keyhistory) # 注意这里的memory_key需要和模板变量名对应 support_chain LLMChain( llmllm, promptprompt, memorymemory, verboseTrue # 打开verbose观察填充后的提示词 ) # 3. 测试 response support_chain.run(我的电脑开机特别慢怎么办) print(response)通过verboseTrue的输出你可以清晰地看到{history}和{input}是如何被实际对话内容替换的从而生成最终发送给模型的提示词。这种可控性是构建高质量、稳定输出应用的基础。5. 赋能AI集成工具与外部知识库如果LLM只是一个知识截止到某个时间点的“闭卷考生”那它的能力是受限的。真正的威力在于让它成为一个“开卷考生”能够调用工具如计算器、搜索引擎、数据库和检索外部知识库如你的公司文档、产品手册。这就是“代理Agent”和“检索增强生成RAG”的核心思想。5.1 为LLM装配工具以联网搜索为例LangChain的Agent框架让LLM能够根据用户的问题自主决定是否需要调用工具、调用哪个工具、以及如何解析工具的返回结果。我们以集成一个“联网搜索”工具为例。首先你需要安装一个模拟搜索的工具包由于真实搜索API需要密钥这里我们用DuckDuckGo搜索作为免费示例但请注意其稳定性和可用性可能因网络而异。同时我们使用更现代的LangChain Agent创建方式。pip install duckduckgo-searchimport os from dotenv import load_dotenv from langchain_community.llms import QianfanLLMEndpoint from langchain.agents import AgentExecutor, create_react_agent from langchain_community.tools import DuckDuckGoSearchRun from langchain import hub # 用于拉取预定义的提示词 load_dotenv() llm QianfanLLMEndpoint( qianfan_akos.getenv(QIANFAN_AK), qianfan_skos.getenv(QIANFAN_SK), modelERNIE-Bot-turbo, temperature0.1, # 降低“创造力”让Agent决策更稳定 ) # 1. 定义工具 search DuckDuckGoSearchRun(nameSearch, description用于搜索互联网上的最新信息。) tools [search] # 2. 从LangChain Hub拉取一个适合ReAct框架的提示词 # ReAct (Reasoning Acting) 是一种让LLM边思考边行动的经典Agent模式 prompt hub.pull(hwchase17/react) # 3. 创建ReAct Agent agent create_react_agent(llm, tools, prompt) # 4. 创建Agent执行器 agent_executor AgentExecutor( agentagent, toolstools, verboseTrue, # 强烈建议打开可以看到Agent的“思考过程” handle_parsing_errorsTrue, # 处理Agent输出解析错误 max_iterations5, # 限制最大迭代次数防止死循环 early_stopping_methodgenerate, # 提前停止策略 ) # 5. 运行Agent try: result agent_executor.invoke({ input: 查询一下今天北京的最高气温是多少度 }) print(f\n最终答案{result[output]}) except Exception as e: print(f执行出错{e})运行这段代码在verboseTrue模式下你会看到类似以下的精彩输出 Entering new AgentExecutor chain... 我需要找到今天北京的最高气温。我应该使用搜索工具来获取最新信息。 行动Search 行动输入今天北京最高气温 观察[搜索返回的结果例如“北京今天晴最高气温25摄氏度...”] 思考根据搜索结果今天北京的最高气温是25摄氏度。 最终答案今天北京的最高气温是25摄氏度。 Finished chain.这就是Agent的魅力LLM自己规划了“需要搜索 - 构造查询词 - 解析搜索结果 - 给出答案”的完整步骤。你可以定义更多工具如查询数据库、调用内部API、执行代码等极大地扩展了LLM的能力边界。重要避坑点工具描述description至关重要LLM根据工具的描述来决定是否以及如何调用它。描述必须清晰、准确说明工具的用途和输入格式。控制迭代与超时务必设置max_iterations如5-10次防止Agent陷入无休止的“思考-行动”循环。handle_parsing_errorsTrue能避免因为输出格式不符合预期而导致整个流程崩溃。Temperature设置对于Agent通常建议设置较低的temperature如0.1以使其决策更加确定和稳定减少随机性带来的不可控行为。5.2 构建私有知识库问答RAG基础实战当问题涉及外部、非公开或最新的文档时我们需要RAG。其核心流程是加载文档 - 分割文本 - 向量化存储 - 检索相关片段 - 组合成提示词交给LLM生成答案。我们以一个简单的本地文本文件问答为例。# 安装处理文档和向量数据库的依赖 pip install langchain-text-splitters chromadb tiktoken # tiktoken用于token计数非必须但推荐。Chromadb是一个轻量级向量数据库。import os from dotenv import load_dotenv from langchain_community.llms import QianfanLLMEndpoint from langchain_community.embeddings import QianfanEmbeddingsEndpoint # 千帆的嵌入模型 from langchain_community.vectorstores import Chroma from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.chains import RetrievalQA from langchain.document_loaders import TextLoader load_dotenv() # 1. 初始化LLM和Embeddings llm QianfanLLMEndpoint( qianfan_akos.getenv(QIANFAN_AK), qianfan_skos.getenv(QIANFAN_SK), modelERNIE-Bot-turbo, ) # 千帆也提供了文本嵌入模型用于将文本转换为向量 embeddings QianfanEmbeddingsEndpoint( qianfan_akos.getenv(QIANFAN_AK), qianfan_skos.getenv(QIANFAN_SK), ) # 2. 加载并分割文档 # 假设你有一个 company_handbook.txt 文件 loader TextLoader(./company_handbook.txt, encodingutf-8) documents loader.load() # 分割文档。chunk_size和chunk_overlap是关键参数。 text_splitter RecursiveCharacterTextSplitter( chunk_size500, # 每个文本块的大小字符数 chunk_overlap50, # 块之间的重叠字符避免上下文断裂 length_functionlen, ) texts text_splitter.split_documents(documents) print(f将文档分割成了 {len(texts)} 个文本块。) # 3. 向量化并存储到向量数据库 # persist_directory 指定持久化目录否则数据只在内存中 persist_directory ./chroma_db vectordb Chroma.from_documents( documentstexts, embeddingembeddings, persist_directorypersist_directory ) vectordb.persist() # 保存到磁盘 print(向量数据库已创建并持久化。) # 4. 创建检索器 retriever vectordb.as_retriever( search_kwargs{k: 3} # 每次检索返回最相关的3个文本块 ) # 5. 创建检索问答链 qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, # 最常用的类型将检索到的所有文档“塞”进提示词 retrieverretriever, return_source_documentsTrue, # 返回源文档便于追溯答案来源 verboseTrue, ) # 6. 进行问答 query 我们公司的年假政策是怎样的 result qa_chain.invoke({query: query}) print(f\n问题{query}) print(f答案{result[result]}) print(\n--- 参考来源 ---) for i, doc in enumerate(result[source_documents]): print(f[片段{i1}]: {doc.page_content[:200]}...) # 打印前200字符RAG实战经验与调优文本分割是艺术chunk_size没有银弹。太小会丢失上下文太大会引入噪声并增加LLM处理负担。对于普通文档500-1000字符是常见起点。对于代码或结构化文本可能需要按行或特定分隔符分割。嵌入模型的选择这里使用了千帆的嵌入模型它和文心一言系列同源在中文语义匹配上通常有较好表现。你也可以选择其他开源嵌入模型如text2vec但需注意其维度与向量数据库的兼容性。检索策略search_kwargs{“k”: 3}表示检索Top 3相关的片段。对于复杂问题可以增加k值。Retriever还支持search_type“mmr”(最大边际相关性)在保证相关性的同时增加多样性。链类型chain_type除了“stuff”简单拼接还有“map_reduce”分别问答再汇总、“refine”迭代精炼等适用于处理非常多的文档块但复杂度更高。来源追溯务必设置return_source_documentsTrue。这对于验证答案准确性、建立用户信任至关重要。在实际产品中你可以在答案后面附上“根据XX文档第Y节”并支持用户点击查看原文。6. 性能、成本与生产化考量当一个原型跑通后要将其变为一个可投入生产环境的服务我们还需要关注性能、成本和稳定性。6.1 流式输出提升用户体验的关键对于需要长时间等待的模型生成流式输出Streaming能极大地改善用户体验让用户看到文字逐个出现而不是干等十几秒后一次性显示全部。千帆的API和LangChain都支持流式响应。以下是实现方法from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler # 初始化支持流式的LLM streaming_llm QianfanLLMEndpoint( qianfan_akos.getenv(QIANFAN_AK), qianfan_skos.getenv(QIANFAN_SK), modelERNIE-Bot-turbo, streamingTrue, # 启用流式 callbacks[StreamingStdOutCallbackHandler()], # 添加流式回调处理器 temperature0.7, ) # 调用时响应会实时打印到标准输出 print(流式问答开始输入‘退出’结束:) while True: user_input input(\n你) if user_input.lower() 退出: break print(AI, end, flushTrue) # 打印前缀但不换行 # invoke方法会触发流式回调 response streaming_llm.invoke(user_input) print() # 换行在Web应用如FastAPI中你可以利用StreamingResponse将生成的内容以SSEServer-Sent Events或类似技术推送给前端。6.2 控制成本与超时设置合理的参数直接调用大模型API成本主要按token消耗计算。我们需要在效果和成本间取得平衡。llm QianfanLLMEndpoint( qianfan_akos.getenv(QIANFAN_AK), qianfan_skos.getenv(QIANFAN_SK), modelERNIE-Bot-turbo, temperature0.1, # 温度越低输出越确定适合事实问答越高越有创意。 top_p0.8, # 核采样与temperature配合影响词的选择范围。 penalty_score1.0, # 重复惩罚1.0降低重复1.0增加重复。 request_timeout60, # 请求超时时间秒根据网络状况调整。 # 注意千帆SDK可能还有 max_tokens, stop 等参数请查阅最新文档 )成本控制实战技巧缓存对相同或相似的查询结果进行缓存可以显著减少对API的调用。LangChain提供了LLMCache组件可以方便地集成。精简上下文如前所述使用ConversationBufferWindowMemory或ConversationSummaryMemory控制输入token数。设置用量告警在千帆控制台设置每日/每月消费额度告警避免意外开销。异步与批处理对于后台任务使用异步调用 (ainvoke) 和非实时处理可以更好地管理并发和资源。6.3 错误处理与重试构建健壮的应用网络请求、模型服务都可能出现暂时性失败。一个健壮的生产应用必须具备错误处理和重试机制。from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type import requests # 定义一个重试装饰器 retry( stopstop_after_attempt(3), # 最多重试3次 waitwait_exponential(multiplier1, min2, max10), # 指数退避等待 retryretry_if_exception_type((requests.exceptions.Timeout, requests.exceptions.ConnectionError)), reraiseTrue, # 重试次数用尽后抛出原始异常 ) def robust_llm_invoke(chain, query): 一个带有重试机制的调用封装 try: response chain.invoke({query: query}) return response except Exception as e: print(f调用发生异常{type(e).__name__}: {e}) # 这里可以根据异常类型进行更精细的处理如令牌超限、频率限制等 if rate limit in str(e).lower(): print(触发频率限制等待更长时间...) # 可以在这里实现更复杂的退避逻辑 raise e # 在关键业务调用处使用封装函数 try: result robust_llm_invoke(qa_chain, 重要的问题) print(result[result]) except Exception as e: print(f所有重试均失败进行降级处理或返回友好错误信息。) # 例如返回一个预设的默认答案 fallback_answer 服务暂时不可用请稍后再试。此外你应该将LLM调用包裹在try...except块中捕获所有可能的异常超时、认证失败、模型过载、输出解析错误等并设计友好的用户降级方案。6.4 监控与日志洞察应用运行状态在生产环境中详细的日志记录至关重要。你需要记录请求与响应至少记录问题的摘要和模型回答的摘要注意隐私可脱敏。Token消耗记录每次调用的输入/输出token数用于成本分析和优化。响应时间监控P95、P99延迟发现性能瓶颈。错误率跟踪不同错误类型如429频率限制、500服务器错误的发生率。你可以使用Python标准的logging模块并集成像prometheus-client这样的库来暴露指标方便使用Grafana等工具进行可视化。import logging import time logging.basicConfig(levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) def logged_invoke(chain, query): start_time time.time() logger.info(fLLM调用开始 - 问题: {query[:50]}...) # 记录前50字符 try: result chain.invoke({query: query}) elapsed time.time() - start_time logger.info(fLLM调用成功 - 耗时: {elapsed:.2f}s - 答案长度: {len(result[result])}) # 可以在这里记录token用量如果API返回 return result except Exception as e: logger.error(fLLM调用失败 - 异常: {e}, exc_infoTrue) raise将上述模块——环境隔离、链与记忆、工具与RAG、性能与健壮性——组合起来你就拥有了一个功能完整、具备生产潜力的LLM应用骨架。从这一个接入点开始你可以根据具体的业务需求无限扩展其能力边界。