AI创想

标题: LangChain [打印本页]
作者: a纯情杜蕾斯    时间: 3 小时前
标题: LangChain
作者：CSDN博客
LangChain 进阶开发者学习笔记

零、环境配置与快速开始

0.1 安装依赖
  1. # 核心库
  2. pip install langchain langchain-core langchain-community
  4. # LLM 提供商（根据需要选择）
  5. pip install langchain-openai      # OpenAI/Azure OpenAI
  6. pip install langchain-anthropic   # Claude
  7. pip install langchain-google-genai # Gemini
  8. pip install langchain-ollama      # 本地模型# 向量数据库（RAG 必需）
  9. pip install chromadb              # 轻量级向量库
  10. pip install faiss-cpu             # Facebook 向量检索# 工具库
  11. pip install langsmith              # 可观测性（可选）
复制代码
0.2 配置 API Key

方式一：环境变量（推荐）
  1. # Linux/MacexportOPENAI_API_KEY="sk-..."exportANTHROPIC_API_KEY="sk-ant-..."# Windows PowerShell$env:OPENAI_API_KEY="sk-..."
复制代码
方式二：代码中配置
  1. import os
  2. os.environ["OPENAI_API_KEY"]="sk-..."# 或直接传参from langchain_openai import ChatOpenAI
  3. llm = ChatOpenAI(api_key="sk-...", model="gpt-4")
复制代码
0.3 验证安装
  1. from langchain_openai import ChatOpenAI
  2. from langchain_core.messages import HumanMessage
  4. # 初始化模型
  5. llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)# 测试调用
  6. response = llm.invoke([HumanMessage(content="你好，请用一句话介绍自己")])print(response.content)
复制代码
执行效果：
  1. 我是 OpenAI 开发的 AI 助手，专注于回答问题、提供信息和协助完成各种任务。
复制代码
0.4 国内环境配置

使用代理：
  1. import os
  2. os.environ["HTTP_PROXY"]="http://127.0.0.1:7890"
  3. os.environ["HTTPS_PROXY"]="http://127.0.0.1:7890"
复制代码
使用国内 API 中转：
  1. llm = ChatOpenAI(
  2.     api_key="your-key",
  3.     base_url="https://api.openai-proxy.com/v1",# 中转地址
  4.     model="gpt-4o-mini")
复制代码
使用本地模型（Ollama）：
  1. # 安装 Ollamacurl-fsSL https://ollama.com/install.sh |sh# 下载模型
  2. ollama pull qwen2.5:7b
复制代码
  1. from langchain_ollama import ChatOllama
  3. llm = ChatOllama(model="qwen2.5:7b", temperature=0)
  4. response = llm.invoke("你好")print(response.content)
复制代码
一、架构全景：LangChain 的设计哲学

1.1 核心定位

LangChain = Spring Boot for LLM Applications
1.2 核心组件对标

LangChain 组件传统后端类比职责
RunnableInterface/抽象类统一调用接口（.invoke(), .stream(), .batch()）
LCEL (Expression Language)管道/中间件链组件编排（prompt | llm | parser）
ChainService 层业务逻辑封装多步骤流程
Agent动态路由控制器根据输入自主决策调用哪些 Tools
Tools外部 API/Repository执行具体操作（搜索、数据库查询、API 调用）
MemorySession/Redis 缓存维护对话上下文
Retriever数据库查询层向量检索/文档召回
二、基础架构：Runnable 与 LCEL

2.1 Runnable：统一调用接口

类比：Java 的 Function<T, R> 或 Spring 的 HandlerInterceptor
所有 LangChain 组件都实现 Runnable 接口，提供标准化调用方式：
  1. from langchain_openai import ChatOpenAI
  2. from langchain_core.messages import HumanMessage
  3. import asyncio
  4. import time
  6. llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)# 1. 同步调用
  7. result = llm.invoke([HumanMessage(content="1+1等于几？")])print(f"同步调用: {result.content}")# 2. 批量调用（一次请求处理多个输入）
  8. batch_results = llm.batch([[HumanMessage(content="Python 之父是谁？")],[HumanMessage(content="Java 之父是谁？")]])for i, res inenumerate(batch_results):print(f"批量结果 {i+1}: {res.content}")# 3. 流式调用（实时输出）print("流式调用: ", end="")for chunk in llm.stream([HumanMessage(content="用一句话介绍 LangChain")]):print(chunk.content, end="", flush=True)print()# 4. 异步调用（高并发场景）asyncdefasync_demo():"""异步调用示例：适合需要同时处理多个请求的场景"""# 4.1 单个异步调用
  9.     result =await llm.ainvoke([HumanMessage(content="什么是异步编程？")])print(f"\n异步调用: {result.content}")# 4.2 并发执行多个请求（性能优势明显）
  10.     start_time = time.time()# 创建多个异步任务
  11.     tasks =[
  12.         llm.ainvoke([HumanMessage(content="什么是 Python？")]),
  13.         llm.ainvoke([HumanMessage(content="什么是 Java？")]),
  14.         llm.ainvoke([HumanMessage(content="什么是 JavaScript？")])]# 并发执行（同时发送3个请求）
  15.     results =await asyncio.gather(*tasks)
  16.    
  17.     elapsed = time.time()- start_time
  18.     print(f"\n并发执行3个请求耗时: {elapsed:.2f}秒")for i, res inenumerate(results):print(f"  结果 {i+1}: {res.content[:50]}...")# 4.3 异步流式输出print("\n异步流式调用: ", end="")asyncfor chunk in llm.astream([HumanMessage(content="用一句话总结 AI 的未来")]):print(chunk.content, end="", flush=True)print()# 运行异步示例
  19. asyncio.run(async_demo())
复制代码
执行效果：
  1. 同步调用: 1+1等于2。
  2. 批量结果 1: Python 之父是 Guido van Rossum。
  3. 批量结果 2: Java 之父是 James Gosling。
  4. 流式调用: LangChain 是一个用于构建基于大语言模型应用的开发框架。
  6. 异步调用: 异步编程是一种编程范式，允许程序在等待某些操作完成时继续执行其他任务，而不是阻塞等待。
  8. 并发执行3个请求耗时: 1.23秒
  9.   结果 1: Python 是一种高级、解释型、通用的编程语言...
  10.   结果 2: Java 是一种面向对象的编程语言，由 Sun Microsystems...
  11.   结果 3: JavaScript 是一种轻量级、解释型的编程语言...
  13. 异步流式调用: AI 的未来将更加智能化、普及化，深度融入人类生活的各个领域。
复制代码
性能对比：
关键方法：
何时使用异步：
2.2 LCEL：管道式组件编排

类比：Spring WebFlux 的 Reactive Streams 或 Unix 管道
  1. from langchain_core.prompts import ChatPromptTemplate
  2. from langchain_core.output_parsers import StrOutputParser
  3. from langchain_openai import ChatOpenAI
  5. # 管道式组装：prompt → llm → parser
  6. chain =(
  7.     ChatPromptTemplate.from_template("将 {text} 翻译成英文")| ChatOpenAI(model="gpt-4o-mini", temperature=0)| StrOutputParser())# 执行
  8. result = chain.invoke({"text":"你好世界"})print(f"翻译结果: {result}")# 批量翻译
  9. batch_results = chain.batch([{"text":"早上好"},{"text":"谢谢"}])for i, res inenumerate(batch_results):print(f"批量翻译 {i+1}: {res}")
复制代码
执行效果：
  1. 翻译结果: Hello, world
  2. 批量翻译 1: Good morning
  3. 批量翻译 2: Thank you
复制代码
核心优势：
2.3 高级编排：并行与条件分支
  1. from langchain_core.runnables import RunnableParallel, RunnableBranch
  2. from langchain_core.prompts import ChatPromptTemplate
  3. from langchain_core.output_parsers import StrOutputParser
  4. from langchain_openai import ChatOpenAI
  6. llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)# 1. 并行执行（类比 CompletableFuture.allOf）
  7. summarize_chain =(
  8.     ChatPromptTemplate.from_template("用一句话总结：{text}")| llm | StrOutputParser())
  9. translate_chain =(
  10.     ChatPromptTemplate.from_template("翻译成英文：{text}")| llm | StrOutputParser())
  11. sentiment_chain =(
  12.     ChatPromptTemplate.from_template("分析情感（正面/负面/中性）：{text}")| llm | StrOutputParser())
  14. parallel_chain = RunnableParallel(
  15.     summary=summarize_chain,
  16.     translation=translate_chain,
  17.     sentiment=sentiment_chain
  18. )
  20. result = parallel_chain.invoke({"text":"今天天气真好，心情很愉快！"})print("并行执行结果:")print(f"  摘要: {result['summary']}")print(f"  翻译: {result['translation']}")print(f"  情感: {result['sentiment']}")# 2. 条件分支（类比 if-else 路由）
  21. chinese_chain = ChatPromptTemplate.from_template("用中文回答：{question}")| llm | StrOutputParser()
  22. english_chain = ChatPromptTemplate.from_template("Answer in English: {question}")| llm | StrOutputParser()
  23. default_chain = ChatPromptTemplate.from_template("回答：{question}")| llm | StrOutputParser()
  25. branch_chain = RunnableBranch((lambda x: x.get("language")=="zh", chinese_chain),(lambda x: x.get("language")=="en", english_chain),
  26.     default_chain  # 默认分支)
  28. result_zh = branch_chain.invoke({"language":"zh","question":"什么是 AI？"})
  29. result_en = branch_chain.invoke({"language":"en","question":"What is AI?"})print(f"\n中文分支: {result_zh}")print(f"英文分支: {result_en}")
复制代码
执行效果：
  1. 并行执行结果:
  2.   摘要: 今天天气好，心情愉快。
  3.   翻译: The weather is really nice today, and I'm in a great mood!
  4.   情感: 正面
  6. 中文分支: AI（人工智能）是指通过计算机系统模拟人类智能的技术...
  7. 英文分支: AI (Artificial Intelligence) refers to the simulation of human intelligence...
复制代码
三、核心特性：Chains、Agents、Memory

3.1 Chain：封装业务流程

类比：Service 层的业务方法
基础 Chain
  1. from langchain_core.prompts import ChatPromptTemplate
  2. from langchain_core.output_parsers import StrOutputParser
  3. from langchain_openai import ChatOpenAI
  5. # 使用 LCEL 构建简单链
  6. prompt = ChatPromptTemplate.from_template("用通俗易懂的语言解释 {concept}")
  7. llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
  8. parser = StrOutputParser()
  10. chain = prompt | llm | parser
  12. # 执行
  13. result = chain.invoke({"concept":"依赖注入"})print(f"解释结果:\n{result}")
复制代码
执行效果：
  1. 解释结果:
  2. 依赖注入就像是你去餐厅吃饭，服务员会把菜端到你面前，而不是让你自己去厨房拿。
  3. 在编程中，依赖注入是指一个对象不自己创建它需要的其他对象，而是由外部传递进来，
  4. 这样可以让代码更灵活、更容易测试。
复制代码
RAG Chain（检索增强生成）

类比：先查数据库，再用查询结果生成响应
  1. from langchain_core.runnables import RunnablePassthrough
  2. from langchain_core.prompts import ChatPromptTemplate
  3. from langchain_core.output_parsers import StrOutputParser
  4. from langchain_openai import ChatOpenAI, OpenAIEmbeddings
  5. from langchain_community.vectorstores import Chroma
  6. from langchain_text_splitters import RecursiveCharacterTextSplitter
  7. from langchain_core.documents import Document
  9. # 1. 准备知识库数据
  10. documents =[
  11.     Document(page_content="向量数据库是专门用于存储和检索高维向量的数据库，常用于 AI 应用。"),
  12.     Document(page_content="Chroma 是一个轻量级的开源向量数据库，支持本地运行。"),
  13.     Document(page_content="RAG（检索增强生成）通过检索相关文档来增强 LLM 的回答准确性。")]# 2. 创建向量存储
  14. vectorstore = Chroma.from_documents(
  15.     documents=documents,
  16.     embedding=OpenAIEmbeddings())
  17. retriever = vectorstore.as_retriever(search_kwargs={"k":2})# 3. 构建 RAG 链defformat_docs(docs):return"\n\n".join([d.page_content for d in docs])
  19. rag_chain =({"context": retriever | format_docs,# 检索并格式化文档"question": RunnablePassthrough()# 透传原始问题}| ChatPromptTemplate.from_template("基于以下上下文回答问题：\n{context}\n\n问题：{question}")| ChatOpenAI(model="gpt-4o-mini", temperature=0)| StrOutputParser())# 4. 执行查询
  20. result = rag_chain.invoke("什么是向量数据库？")print(f"RAG 回答:\n{result}")
复制代码
执行效果：
  1. RAG 回答:
  2. 向量数据库是专门用于存储和检索高维向量的数据库系统，常用于 AI 应用场景。
  3. 例如 Chroma 就是一个轻量级的开源向量数据库，支持本地运行，非常适合快速原型开发。
复制代码
关键点：
3.2 Agent：自主决策执行器

类比：动态路由 + 策略模式
Agent 会根据用户输入自主决定调用哪些 Tools，类似于：
  1. // 伪代码if(需要搜索){
  2.     result = searchTool.execute();}elseif(需要计算){
  3.     result = calculatorTool.execute();}
复制代码
ReAct Agent（推理 + 行动）

注意：Tool的函数中，必须至少要有一个入参作为占位，即使在函数中没有使用
  1. from langchain.agents import AgentExecutor, create_react_agent
  2. from langchain_core.tools import Tool
  3. from langchain_core.prompts import PromptTemplate
  4. from langchain_openai import ChatOpenAI
  6. # 1. 定义工具defsearch_func(query:str)->str:"""模拟搜索功能"""
  7.     knowledge_base ={"2022世界杯":"2022年世界杯冠军是阿根廷","Python":"Python 是一种高级编程语言",}for key, value in knowledge_base.items():if key in query:return value
  8.     returnf"未找到关于 '{query}' 的信息"defcalculator_func(expression:str)->str:"""计算器工具"""try:
  9.         result =eval(expression)returnf"计算结果: {result}"except Exception as e:returnf"计算错误: {str(e)}"
  11. search_tool = Tool(
  12.     name="搜索",
  13.     func=search_func,
  14.     description="用于查询最新信息和知识，输入应该是搜索关键词")
  16. calculator_tool = Tool(
  17.     name="计算器",
  18.     func=calculator_func,
  19.     description="用于执行数学计算，输入应该是数学表达式，例如 '2+3*4'")
  21. tools =[search_tool, calculator_tool]# 2. 创建 Agent（使用 ReAct 模板）from langchain import hub
  22. prompt = hub.pull("hwchase17/react")
  24. llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
  25. agent = create_react_agent(llm, tools, prompt)# 3. 创建执行器
  26. agent_executor = AgentExecutor(
  27.     agent=agent,
  28.     tools=tools,
  29.     verbose=True,# 打印推理过程
  30.     max_iterations=5,
  31.     handle_parsing_errors=True)# 4. 执行任务print("=== 任务1: 需要搜索 ===")
  32. result1 = agent_executor.invoke({"input":"2022年世界杯冠军是谁？"})print(f"最终答案: {result1['output']}\n")print("=== 任务2: 需要计算 ===")
  33. result2 = agent_executor.invoke({"input":"计算 (25 + 15) * 3"})print(f"最终答案: {result2['output']}\n")print("=== 任务3: 需要多步推理 ===")
  34. result3 = agent_executor.invoke({"input":"如果一本书30元，我买5本需要多少钱？"})print(f"最终答案: {result3['output']}")
复制代码
执行效果：
  1. === 任务1: 需要搜索 ===
  3. > Entering new AgentExecutor chain...
  4. Thought: 我需要搜索2022年世界杯的信息
  5. Action: 搜索
  6. Action Input: 2022世界杯
  7. Observation: 2022年世界杯冠军是阿根廷
  8. Thought: 我现在知道答案了
  9. Final Answer: 2022年世界杯冠军是阿根廷
  11. > Finished chain.
  12. 最终答案: 2022年世界杯冠军是阿根廷
  14. === 任务2: 需要计算 ===
  16. > Entering new AgentExecutor chain...
  17. Thought: 这是一个数学计算问题
  18. Action: 计算器
  19. Action Input: (25 + 15) * 3
  20. Observation: 计算结果: 120
  21. Thought: 我得到了计算结果
  22. Final Answer: 120
  24. > Finished chain.
  25. 最终答案: 120
  27. === 任务3: 需要多步推理 ===
  29. > Entering new AgentExecutor chain...
  30. Thought: 需要计算5本书的总价
  31. Action: 计算器
  32. Action Input: 30 * 5
  33. Observation: 计算结果: 150
  34. Thought: 我得到了答案
  35. Final Answer: 买5本书需要150元
  37. > Finished chain.
  38. 最终答案: 买5本书需要150元
复制代码
执行流程：
自定义 Tool
  1. from langchain_core.tools import tool
  3. @tooldefquery_database(sql:str)->str:"""执行 SQL 查询并返回结果
  4.    
  5.     Args:
  6.         sql: SQL 查询语句
  7.    
  8.     Returns:
  9.         查询结果的字符串表示
  10.     """# 模拟数据库查询
  11.     mock_db ={"SELECT * FROM users":"用户列表: [张三, 李四, 王五]","SELECT COUNT(*) FROM orders":"订单总数: 1523"}return mock_db.get(sql,f"执行查询: {sql}")# 集成到 Agent
  12. tools =[query_database, search_tool, calculator_tool]
  13. agent = create_react_agent(llm, tools, prompt)
  14. agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
  16. result = agent_executor.invoke({"input":"查询用户列表"})print(f"结果: {result['output']}")
复制代码
执行效果：
  1. > Entering new AgentExecutor chain...
  2. Thought: 需要查询数据库
  3. Action: query_database
  4. Action Input: SELECT * FROM users
  5. Observation: 用户列表: [张三, 李四, 王五]
  6. Final Answer: 用户列表包括：张三、李四、王五
  8. > Finished chain.
  9. 结果: 用户列表包括：张三、李四、王五
复制代码
3.3 Memory：对话上下文管理

类比：Session 管理 + Redis 缓存
ConversationBufferMemory（完整历史）
  1. from langchain.memory import ConversationBufferMemory
  2. from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
  3. from langchain_core.output_parsers import StrOutputParser
  4. from langchain_openai import ChatOpenAI
  6. # 1. 创建 Memory
  7. memory = ConversationBufferMemory(
  8.     memory_key="chat_history",
  9.     return_messages=True)# 2. 构建带 Memory 的 Chain
  10. llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
  12. prompt = ChatPromptTemplate.from_messages([("system","你是一个友好的助手，能记住之前的对话内容"),
  13.     MessagesPlaceholder(variable_name="chat_history"),# 历史消息占位符("user","{input}")])
  15. chain = prompt | llm | StrOutputParser()# 3. 对话循环defchat_with_memory(user_input:str):# 加载历史
  16.     history = memory.load_memory_variables({})# 执行对话
  17.     response = chain.invoke({"input": user_input,"chat_history": history.get("chat_history",[])})# 保存到 Memory
  18.     memory.save_context({"input": user_input},{"output": response})return response
  20. # 测试对话print("对话1:", chat_with_memory("我叫张三"))print("对话2:", chat_with_memory("我喜欢编程"))print("对话3:", chat_with_memory("我叫什么名字？"))print("对话4:", chat_with_memory("我的爱好是什么？"))# 查看完整历史print("\n完整对话历史:")print(memory.load_memory_variables({}))
复制代码
执行效果：
  1. 对话1: 你好张三！很高兴认识你。
  2. 对话2: 编程是一项很有趣的技能！你主要使用哪种编程语言呢？
  3. 对话3: 你叫张三。
  4. 对话4: 你的爱好是编程。
  6. 完整对话历史:
  7. {'chat_history': [
  8.     HumanMessage(content='我叫张三'),
  9.     AIMessage(content='你好张三！很高兴认识你。'),
  10.     HumanMessage(content='我喜欢编程'),
  11.     AIMessage(content='编程是一项很有趣的技能！你主要使用哪种编程语言呢？'),
  12.     HumanMessage(content='我叫什么名字？'),
  13.     AIMessage(content='你叫张三。'),
  14.     HumanMessage(content='我的爱好是什么？'),
  15.     AIMessage(content='你的爱好是编程。')
  16. ]}
复制代码
ConversationSummaryMemory（摘要压缩）

类比：定期归档旧数据，只保留摘要
  1. from langchain.memory import ConversationSummaryMemory
  3. # 创建摘要 Memory
  4. summary_memory = ConversationSummaryMemory(
  5.     llm=ChatOpenAI(model="gpt-4o-mini"),
  6.     max_token_limit=100# 超过限制自动摘要)# 模拟长对话
  7. conversations =[("我叫张三，是一名软件工程师","你好张三！很高兴认识你这位软件工程师。"),("我在北京工作","北京是个很棒的城市！"),("我喜欢 Python 和 Java","这两门语言都很实用！"),("我最近在学习 LangChain","LangChain 是个很强大的框架！")]for user_msg, ai_msg in conversations:
  8.     summary_memory.save_context({"input": user_msg},{"output": ai_msg})# 查看摘要（而非完整历史）print("对话摘要:")print(summary_memory.load_memory_variables({}))
复制代码
执行效果：
  1. 对话摘要:
  2. {'history': '张三是一名在北京工作的软件工程师，喜欢 Python 和 Java，最近在学习 LangChain 框架。'}
复制代码
持久化 Memory（使用文件存储）
  1. from langchain.memory import FileChatMessageHistory, ConversationBufferMemory
  3. # 使用文件持久化（生产环境建议用 Redis/PostgreSQL）
  4. message_history = FileChatMessageHistory("chat_history.json")
  6. persistent_memory = ConversationBufferMemory(
  7.     chat_memory=message_history,
  8.     return_messages=True)# 保存对话
  9. persistent_memory.save_context({"input":"我的用户ID是12345"},{"output":"已记录您的用户ID"})# 重启程序后仍可读取print("持久化历史:", persistent_memory.load_memory_variables({}))
复制代码
执行效果：
  1. 持久化历史: {'chat_history': [
  2.     HumanMessage(content='我的用户ID是12345'),
  3.     AIMessage(content='已记录您的用户ID')
  4. ]}
  6. # chat_history.json 文件内容:
  7. [
  8.     {"type": "human", "data": {"content": "我的用户ID是12345"}},
  9.     {"type": "ai", "data": {"content": "已记录您的用户ID"}}
  10. ]
复制代码
四、高级用法：自定义与优化

4.1 自定义 Agent 逻辑
  1. from langchain.agents import BaseSingleActionAgent
  2. from typing import List, Tuple, Any
  4. classCustomAgent(BaseSingleActionAgent):"""自定义 Agent：实现特定业务逻辑"""defplan(
  5.         self,
  6.         intermediate_steps: List[Tuple[Any,str]],**kwargs
  7.     )-> Any:"""决策逻辑：选择下一步行动"""# 自定义决策算法if"搜索"in kwargs["input"]:return AgentAction(tool="search", tool_input=kwargs["input"])else:return AgentFinish(return_values={"output":"无需工具"})asyncdefaplan(self,*args,**kwargs):"""异步版本"""return self.plan(*args,**kwargs)
复制代码
4.2 复杂 RAG 链路

Multi-Query RAG（多查询召回）
  1. from langchain.retrievers import MultiQueryRetriever
  3. # 自动生成多个查询变体
  4. retriever = MultiQueryRetriever.from_llm(
  5.     retriever=base_retriever,
  6.     llm=ChatOpenAI())# 输入："什么是 RAG？"# 自动生成：["RAG 是什么", "检索增强生成的定义", "RAG 技术原理"]
复制代码
Re-ranking（重排序）
  1. from langchain.retrievers import ContextualCompressionRetriever
  2. from langchain.retrievers.document_compressors import CohereRerank
  4. compressor = CohereRerank(model="rerank-english-v2.0")
  6. compression_retriever = ContextualCompressionRetriever(
  7.     base_compressor=compressor,
  8.     base_retriever=retriever
  9. )
复制代码
4.3 流式输出（实时响应）
  1. from langchain_core.prompts import ChatPromptTemplate
  2. from langchain_core.output_parsers import StrOutputParser
  3. from langchain_openai import ChatOpenAI
  4. import time
  6. llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)# 构建链
  7. chain =(
  8.     ChatPromptTemplate.from_template("写一首关于 {topic} 的五言绝句")| llm
  9.     | StrOutputParser())# 1. 同步流式输出print("=== 同步流式输出 ===")for chunk in chain.stream({"topic":"春天"}):print(chunk, end="", flush=True)
  10.     time.sleep(0.05)# 模拟打字机效果print("\n")# 2. 异步流式输出（高并发场景）import asyncio
  12. asyncdefasync_stream_demo():print("=== 异步流式输出 ===")asyncfor chunk in chain.astream({"topic":"编程"}):print(chunk, end="", flush=True)await asyncio.sleep(0.05)print("\n")# 运行异步示例
  13. asyncio.run(async_stream_demo())# 3. 带 Token 使用统计的流式输出from langchain_core.callbacks import StreamingStdOutCallbackHandler
  15. llm_with_callback = ChatOpenAI(
  16.     model="gpt-4o-mini",
  17.     streaming=True,
  18.     callbacks=[StreamingStdOutCallbackHandler()])
  20. chain_with_callback =(
  21.     ChatPromptTemplate.from_template("用一句话介绍 {concept}")| llm_with_callback
  22.     | StrOutputParser())print("=== 带回调的流式输出 ===")
  23. result = chain_with_callback.invoke({"concept":"LangChain"})print(f"\n完整结果: {result}")
复制代码
执行效果：
  1. === 同步流式输出 ===
  2. 春风拂柳绿，
  3. 花开满园香。
  4. 燕归巢中语，
  5. 万物复苏忙。
  7. === 异步流式输出 ===
  8. 代码如诗行，
  9. 逻辑织梦想。
  10. 键盘敲击声，
  11. 创造新世界。
  13. === 带回调的流式输出 ===
  14. LangChain 是一个用于构建基于大语言模型应用的开发框架。
  15. 完整结果: LangChain 是一个用于构建基于大语言模型应用的开发框架。
复制代码
4.4 错误处理与重试
  1. from langchain_core.runnables import RunnableRetry
  3. # 自动重试（类比 Spring Retry）
  4. chain_with_retry = chain.with_retry(
  5.     stop_after_attempt=3,
  6.     wait_exponential_multiplier=1,
  7.     wait_exponential_max=10)# 降级处理
  8. chain_with_fallback = chain.with_fallbacks([backup_chain],
  9.     exceptions_to_handle=(RateLimitError,))
复制代码
五、最佳实践

5.1 性能优化

5.2 可观测性
  1. from langsmith import Client
  3. # 集成 LangSmith 追踪
  4. client = Client()# 自动记录所有调用链路with client.trace("my_chain"):
  5.     result = chain.invoke(input)
复制代码
5.3 安全性

六、核心概念速查表

概念用途代码示例
Runnable统一接口component.invoke(input)
LCEL管道编排prompt | llm | parser
Chain业务流程LLMChain(llm, prompt)
Agent自主决策create_react_agent(llm, tools)
Tool外部能力@tool def func(): ...
Memory上下文ConversationBufferMemory()
Retriever文档召回vectorstore.as_retriever()

原文地址：https://blog.csdn.net/hsnsnusjs/article/details/160501263



欢迎光临 AI创想 (http://llms-ai.com/) Powered by Discuz! X3.4