LangChain 深入

这里需要装什么包什么依赖 我就不再一一赘述了 大家可以先看上一篇 《Langchain 浅出》

那么如果出现缺失的依赖怎么办 ？简单 缺什么装什么

目录

• 1、Python 依赖安装
• 2、词工程最佳实践
• 3、性能优化技巧
• 4、常见问题与解决方案
• 5、调试和错误处理
• 6、生产环境最佳实践

想了想还是给补一份基础的依赖吧 ，至于为什么，我也不知道 但是我还是补上了

另外

本章篇幅比较密的代码示例需要个人花点时间理解和消化有问题可以在评论区交流

Python 依赖安装（全套覆盖）

核心依赖（所有 demo 都需要）

1pip install --upgrade pip setuptools wheel
2pip install python-dotenv           # 环境变量管理
3pip install langchain==1.0.*       # LangChain 1.x 核心
4pip install langchain-ollama       # Ollama LLM 支持
5pip install ollama                  # Ollama 官方客户端（模型自检/拉取）
6

向量数据库与文档检索（文档助手/检索 demo）

1pip install langchain-chroma        # Chroma 向量数据库
2pip install langchain-community     # 社区扩展功能（额外工具、插件）
3pip install chromadb                # Chroma 官方依赖
4

文档加载（PDF / Word / 通用文档）

1pip install unstructured           # 文本/结构化文档加载
2pip install pypdf                   # PDF 文档解析
3pip install python-docx             # Word 文档解析
4

Embeddings / NLP（可选替代 Ollama Embedding）

1pip install sentence-transformers   # 文本向量生成，可选
2

词工程最佳实践

1flowchart TD
2    A[开始] --> B["定义示例数据<br>examples"]
3    B --> C["创建PromptTemplate<br>example_prompt"]
4    C --> D["设置LengthBasedExampleSelector<br>selector"]
5    D --> E["构建FewShotPromptTemplate<br>few_shot"]
6    E --> F["初始化ChatOllama模型"]
7    F --> G["遍历输入情绪列表"]
8    G --> H["格式化提示词<br>few_shot.format()"]
9    H --> I["调用模型生成回复<br>llm.invoke()"]
10    I --> J["打印回复结果"]
11    J --> K{还有更多情绪?}
12    K -->|是| G
13    K -->|否| END(("END"))
14
15

创建文件 advanced_prompting.py：

1"""
2高级提示词技巧（基于 LangChain 1.0 + LangGraph）
3"""
4
5from langchain_ollama import ChatOllama
6from langchain_core.prompts import (
7    ChatPromptTemplate, # 提示词模板
8    SystemMessagePromptTemplate,# 系统消息模板
9    HumanMessagePromptTemplate,# 角色消息模板
10    PromptTemplate, # 单个示例格式
11    FewShotPromptTemplate # 少样本学习示例
12)
13from langchain_core.example_selectors import LengthBasedExampleSelector
14# from dotenv import load_dotenv 还是和之前一样 因为我们本地使用的ollama基本都是默认的配置所以暂时不加载配置文件 
15
16# load_dotenv()
17
18
19def few_shot_learning():
20    """少样本学习：通过示例教模型执行任务（LangChain 1.0 风格）"""
21
22    examples = [
23        {"input": "开心", "output": "😊 今天心情真好！"},
24        {"input": "难过", "output": "😢 有些失落，不过会慢慢好起来。"},
25        {"input": "兴奋", "output": "🎉 太棒了！令人振奋的消息！"}
26    ]
27
28    # 单个示例格式
29    example_prompt = PromptTemplate(
30        template="情绪：{input}\n回复：{output}",
31        input_variables=["input", "output"],
32    )
33
34    # 示例选择器（自动选择合适数量）
35    selector = LengthBasedExampleSelector(
36        examples=examples,
37        example_prompt=example_prompt,
38        max_length=100
39    )
40
41    few_shot = FewShotPromptTemplate(
42        example_selector=selector,
43        example_prompt=example_prompt,
44        prefix="你是一个情感友好的 AI 助手，会用 emoji 回复：\n\n示例：",
45        suffix="\n情绪：{input}\n回复：",
46        input_variables=["input"]
47    )
48
49    llm = ChatOllama(model="qwen3-coder:30b", temperature=0.7)
50
51    emotions = ["激动", "焦虑", "感动"]
52
53    print("\n=== 少样本学习示例 ===")
54    for emotion in emotions:
55        prompt = few_shot.format(input=emotion)
56        res = llm.invoke(prompt)
57        print(f"\n情绪：{emotion}")
58        print(f"AI 回复：{res.content}")
59
60
61def role_based_prompting():
62    """基于角色的提示词（新版 lc3 PromptTemplate）"""
63
64    system_template = """
65你是一名{role}。
66特点：{characteristics}
67回答风格：{style}
68"""
69
70    system_msg = SystemMessagePromptTemplate.from_template(system_template)
71    human_msg = HumanMessagePromptTemplate.from_template("{question}")
72
73    prompt = ChatPromptTemplate.from_messages([system_msg, human_msg])
74
75    llm = ChatOllama(model="qwen3-coder:30b", temperature=0.6)
76
77    roles = [
78        {
79            "role": "资深软件架构师",
80            "characteristics": "熟悉微服务、性能优化、分布式架构",
81            "style": "严谨、专业、会给出架构图示建议",
82            "question": "如何设计一个高并发的电商系统？"
83        },
84        {
85            "role": "耐心的编程老师",
86            "characteristics": "善于用比喻解释复杂概念",
87            "style": "温和、逐步讲解、举例子",
88            "question": "什么是类和对象？"
89        }
90    ]
91
92    print("\n=== 基于角色的提示词示例 ===")
93    for config in roles:
94        msgs = prompt.format_messages(**config)
95        res = llm.invoke(msgs)
96        print(f"\n角色：{config['role']}")
97        print(f"回答：{res.content}")
98
99
100def chain_of_thought_prompting():
101    """思维链提示（手动 CoT）"""
102
103    llm = ChatOllama(model="qwen3-coder:30b", temperature=0)
104
105    normal = "商店有23个苹果，卖出17个，又进了30个，现在有多少？"
106
107    cot = """
108商店有23个苹果，卖出17个，又进30个。
109
110请按照步骤推理：
1111. 卖出后剩余多少
1122. 再加上进货数量
1133. 计算最终结果
114"""
115
116    print("\n=== 思维链对比 ===")
117
118    print("\n【普通提示】")
119    print(llm.invoke(normal).content)
120
121    print("\n【思维链 CoT】")
122    print(llm.invoke(cot).content)
123
124
125if __name__ == "__main__":
126    few_shot_learning()
127    role_based_prompting()
128    chain_of_thought_prompting()
129
130

输出结果：

太长这里就放个截图示意吧

性能优化技巧

1flowchart TD
2    开始[开始程序] --> 功能选择
3
4    功能选择 --> 缓存示例
5    功能选择 --> 批处理示例
6    功能选择 --> 流式输出示例
7
8    %% 缓存示例
9    缓存示例 --> 创建缓存
10    创建缓存 --> 初始化LLM
11    初始化LLM --> 第一次调用
12    第一次调用 --> 模型推理
13    模型推理 --> 存入缓存
14    初始化LLM --> 第二次调用
15    第二次调用 --> 缓存命中
16    缓存命中 --> 缓存完成
17
18    %% 批处理示例
19    批处理示例 --> 准备问题列表
20    准备问题列表 --> 循环调用
21    循环调用 --> 记录耗时1
22    准备问题列表 --> 批量调用
23    批量调用 --> 记录耗时2
24    记录耗时2 --> 批处理完成
25
26    %% 流式输出示例
27    流式输出示例 --> 初始化LLM流
28    初始化LLM流 --> 调用流方法
29    调用流方法 --> 接收分块
30    接收分块 --> 流处理完成
31
32    缓存完成 --> 结束
33    批处理完成 --> 结束
34    流处理完成 --> 结束
35
36

创建文件 optimization_tips.py：

1"""
2LangChain 1.x 性能优化技巧示例 —— 最新写法（2025）
3"""
4
5import time
6from dotenv import load_dotenv
7load_dotenv()
8
9# LangChain 1.x 最新模块
10from langchain_core.caches import InMemoryCache
11
12
13# Ollama LLM
14from langchain_ollama import ChatOllama
15
16
17def caching_example():
18    """使用缓存减少重复调用"""
19
20    print("=" * 50)
21    print("缓存优化示例（LangChain 1.x）")
22    print("=" * 50)
23
24    # 设置全局缓存（官方推荐方式）
25    cache = InMemoryCache()
26
27    llm = ChatOllama(model="qwen3-coder:30b", temperature=0.7,cache=cache)
28
29    question = "什么是机器学习？"
30
31    # 第一次调用（非缓存）
32    print(f"\n问题：{question}")
33    print("第一次调用（未缓存）...")
34    start = time.time()
35    response1 = llm.invoke(question)
36    time1 = time.time() - start
37    print(f"耗时：{time1:.2f} 秒")
38    print(f"答案：{response1.content[:100]}...")
39
40    # 第二次调用（缓存命中）
41    print("\n第二次调用（已缓存）...")
42    start = time.time()
43    response2 = llm.invoke(question)
44    time2 = time.time() - start
45    print(f"耗时：{time2:.2f} 秒")
46    if time2 == 0:
47     print("🚀 速度提升：∞（缓存命中，耗时约 0 秒）")
48    else:
49        print(f"🚀 速度提升：{time1/time2:.1f} 倍")
50
51
52def batch_processing():
53    """批量处理以减少网络调用"""
54
55    print("\n" + "=" * 50)
56    print("批量处理示例（LangChain 1.x）")
57    print("=" * 50)
58
59    llm = ChatOllama(model="qwen3-coder:30b", temperature=0.7)
60
61    questions = [
62        "Python 是什么？",
63        "Java 是什么？",
64        "JavaScript 是什么？"
65    ]
66
67    # 方式1：逐个处理
68    print("\n方式1：逐个调用 invoke()")
69    start = time.time()
70    for q in questions:
71        llm.invoke(q)
72    time1 = time.time() - start
73    print(f"总耗时：{time1:.2f} 秒")
74
75    # 方式2：批量处理（LangChain 1.x 推荐）
76    print("\n方式2：使用 llm.batch()")
77    start = time.time()
78    responses = llm.batch(questions)
79    time2 = time.time() - start
80    if time2 == 0:
81        print("🚀 速度提升：∞（缓存命中，耗时约 0 秒）")
82    else:
83        print(f"🚀 速度提升：{time1/time2:.1f} 倍")
84
85
86def streaming_output():
87    """流式输出，提升用户体验"""
88
89    print("\n" + "=" * 50)
90    print("流式输出示例（LangChain 1.x）")
91    print("=" * 50)
92
93    # 流式输出无需 streaming=True
94    llm = ChatOllama(
95        model="qwen3-coder:30b",
96        temperature=0.7,
97    )
98
99    print("\n生成中：", end="", flush=True)
100
101    # LangChain 1.x 官方写法：llm.stream()
102    for chunk in llm.stream("请用100字介绍人工智能的发展历史"):
103        print(chunk.content, end="", flush=True)
104
105    print("\n")
106
107
108if __name__ == "__main__":
109    caching_example()
110    batch_processing()
111    streaming_output()
112
113

输出结果:

1==================================================
2缓存优化示例（LangChain 1.x）
3==================================================
4
5问题：什么是机器学习？
6第一次调用（未缓存）...
7耗时：1.35 秒
8答案：机器学习是人工智能的一个重要分支，它让计算机能够在不被明确编程的情况下，通过分析大量数据来自动学习和改进。
9
10## 核心概念
11机器学习让计算机从经验中学习规律，并利用这些规律对新的、未见过的数据进行预...
12
13第二次调用（已缓存）...
14耗时：0.00 秒
15🚀 速度提升：∞（缓存命中，耗时约 0 秒）
16
17==================================================
18批量处理示例（LangChain 1.x）
19==================================================
20
21方式1：逐个调用 invoke()
22总耗时：4.56 秒
23
24方式2：使用 llm.batch()
25🚀 速度提升：1.0 倍
26
27==================================================
28流式输出示例（LangChain 1.x）
29==================================================
30
31生成中：人工智能发展可追溯至1950年代。1956年达特茅斯会议正式确立AI概念。经历了多次起伏：1970年代专家系统兴起，1980年代遭遇"AI寒冬"，1990年代机器学习兴起。21世纪后，大数据、深度学习技术突破，AI迎来快速发展期，在图像识
32别、自然语言处理等领域取得重大进展，正深刻改变着人类生活。
33
34(venv) PS D:\softwares\langchin_learning> 
35

常见问题与解决方案（1.0 版本）

1. 模型初始化与 API Key

问题：模型无法调用或报错

解决方案：

• 对 OpenAI：检查 API Key 是否正确，确保环境变量已加载

1import os  
2from dotenv import load_dotenv  
3load_dotenv()  
4api_key = os.getenv("OPENAI_API_KEY")  

• 对 ChatOllama / 本地 Ollama 模型：无需网络 API，只要 Ollama Server 已启动即可

1ollama list  # 检查模型是否可用  

• 确保 虚拟环境正确激活，并安装最新依赖：

1pip install -U langchain langchain-ollama python-dotenv  

2. 依赖包版本冲突

问题：ImportError 或模块版本不兼容

解决方案：

• 使用虚拟环境隔离项目依赖（venv / conda / poetry）
• 升级至最新版本：

1pip install --upgrade langchain langchain-ollama python-dotenv  

• 确保 Python 版本 3.9 以上

3. 网络与本地服务连接问题

问题：模型无法访问

解决方案：

• OpenAI：确保网络可用，必要时配置代理
• ChatOllama（本地模型）：
  • Ollama Server 必须启动
  • 使用命令检查模型列表：

  1ollama list  

  • 测试本地端口：

  1curl http://localhost:11434  

4. Token 限制或显存问题

问题：模型响应超时或显存不足

解决方案：

• OpenAI：注意速率限制（RateLimit），可降低请求频率或批量调用
• 本地 Ollama：
  • 没有调用次数限制，但受显卡显存影响
  • 可选择小模型，如：

  1Qwen2.5:3B < Qwen2.5:7B < Qwen3-coder:30B  

• 对于超长 Prompt：建议分段或使用 llm.batch() 批量调用

5. 流式输出与回调问题

问题：流式输出不生效或回调不触发

解决方案：

• 流式输出：
  • 最新 LangChain 1.x 不再使用 streaming=True 参数
  • 正确方法：

  1for chunk in llm.stream("你的 Prompt"):  
  2    print(chunk.content, end="")  

• 回调：
  • 使用 langchain_core.callbacks.BaseCallbackHandler 编写自定义回调
  • 确保在 LLM 或 Runnable 初始化时传入：

  1from langchain_core.callbacks import CallbackManager, BaseCallbackHandler  
  2class MyHandler(BaseCallbackHandler):  
  3    def on_llm_new_token(self, token):  
  4        print(token, end="")  
  5callback_manager = CallbackManager([MyHandler()])  
  6llm = ChatOllama(model="Qwen2.5:7b", callback_manager=callback_manager)  

6. 其他常见问题

• 缓存未生效：确保使用最新 InMemoryCache 或 PersistentCache，并传入 LLM：

1from langchain_core.caches import InMemoryCache  
2llm = ChatOllama(model="Qwen2.5:7b", cache=InMemoryCache())  

• 批量调用报错：检查输入是 list[str]，使用 .batch() 方法
• 超长输出截断：可使用 max_tokens 参数或分段请求

调试和错误处理

创建文件 debugging_guide.py：

1"""
2LangChain 1.x 调试与错误处理示例 —— 最新写法（2025）
3"""
4
5import logging
6from dotenv import load_dotenv
7load_dotenv()
8
9# 最新 LangChain 1.x 模块
10from langchain_ollama import ChatOllama
11from langchain_core.callbacks import StdOutCallbackHandler
12
13# 配置日志
14logging.basicConfig(level=logging.INFO)
15logger = logging.getLogger(__name__)
16
17
18def verbose_mode_example():
19    """使用回调追踪详细执行过程（替代旧的 LLMChain + verbose）"""
20    
21    print("=" * 50)
22    print("Verbose 模式示例")
23    print("=" * 50)
24    
25    # 初始化 Ollama LLM
26    llm = ChatOllama(model="qwen3-coder:30b", temperature=0.7)
27    
28    # 回调处理器打印详细信息
29    handler = StdOutCallbackHandler()
30    
31    # 直接调用 LLM 并传入回调
32    response = llm.invoke(
33        "请用一句话介绍量子计算",
34        config={"callbacks": [handler]}
35    )
36    
37    print(f"\n最终结果：{response.content}")
38
39
40def callback_example():
41    """使用回调处理器追踪 LLM 调用"""
42    
43    print("\n" + "=" * 50)
44    print("回调处理器示例")
45    print("=" * 50)
46    
47    llm = ChatOllama(model="qwen3-coder:30b", temperature=0.7)
48    
49    handler = StdOutCallbackHandler()
50    
51    response = llm.invoke(
52        "什么是区块链？",
53        config={"callbacks": [handler]}
54    )
55    
56    print(f"\n答案：{response.content}")
57
58
59def error_handling_example():
60    """错误处理最佳实践"""
61    
62    print("\n" + "=" * 50)
63    print("错误处理示例")
64    print("=" * 50)
65    
66    llm = ChatOllama(model="qwen3-coder:30b", temperature=0.7)
67    
68    try:
69        response = llm.invoke("测试问题")
70        print(f"成功：{response.content}")
71        
72    except Exception as e:
73        logger.error(f"调用 LLM 出错: {type(e).__name__}")
74        logger.error(f"错误详情: {str(e)}")
75        
76        err_msg = str(e).lower()
77        if "rate_limit" in err_msg:
78            print("触发速率限制，请稍后重试")
79        elif "authentication" in err_msg:
80            print("API 密钥错误，请检查配置")
81        elif "timeout" in err_msg:
82            print("请求超时，请检查网络")
83        else:
84            print(f"未知错误：{e}")
85
86
87def token_counting():
88    """Token 计数示例（仅适用于 OpenAI API）"""
89    
90    print("\n" + "=" * 50)
91    print("Token 计数示例")
92    print("=" * 50)
93    
94    llm = ChatOllama(model="qwen3-coder:30b", temperature=0.7)
95    
96    # Ollama 本地模型无法使用 get_openai_callback
97    print("⚠ Ollama 本地模型不支持 OpenAI Token 计数，此示例仅供 OpenAI API 使用")
98    
99    # OpenAI 可用写法（保留示例）
100    """
101    from langchain_core.callbacks import get_openai_callback
102    with get_openai_callback() as cb:
103        response = llm.invoke("请详细介绍 Python 编程语言的特点和应用场景")
104        print(f"答案：{response.content}")
105        print(f"总 Token：{cb.total_tokens}, 提示词 Token：{cb.prompt_tokens}, 响应 Token：{cb.completion_tokens}")
106        print(f"总成本：${cb.total_cost:.4f}")
107    """
108
109
110if __name__ == "__main__":
111    verbose_mode_example()
112    callback_example()
113    error_handling_example()
114    token_counting()
115
116

输出结果:

1==================================================
2Verbose 模式示例
3==================================================
4INFO:httpx:HTTP Request: POST http://127.0.0.1:11434/api/chat "HTTP/1.1 200 OK"
5
6最终结果：量子计算是一种利用量子比特的叠加和纠缠特性进行并行计算的新型计算方式，能够在某些特定问题上实现指数级的计算速度提升。
7
8==================================================
9回调处理器示例
10==================================================
11INFO:httpx:HTTP Request: POST http://127.0.0.1:11434/api/chat "HTTP/1.1 200 OK"
12
13答案：区块链是一种**分布式数据库技术**，它将数据存储在按时间顺序链接的区块中，每个区块都包含前一个区块的加密哈希值，形成一个不可篡改的链式结构。
14
15## 核心特征
16
17**1. 去中心化**
18- 没有单一控制者
19- 多个节点共同维护网络
20
21**2. 不可篡改**
22- 一旦数据写入就很难修改
23- 修改需要网络中大多数节点同意
24
25**3. 透明可追溯**
26- 所有交易记录公开可见
27- 可以追踪数据的完整历史
28
29## 工作原理
30
311. **区块创建**：新数据被打包成区块
322. **加密验证**：通过密码学算法验证区块
333. **网络共识**：多个节点确认区块有效性
344. **链式连接**：新区块链接到现有链上
35
36## 主要应用
37- **加密货币**：比特币、以太坊等
38- **智能合约**：自动执行的合约协议
39- **供应链管理**：商品溯源追踪
40- **数字身份**：身份认证和管理
41- **医疗记录**：安全存储和共享
42
43区块链技术正在多个领域发挥重要作用，为数据安全和信任建立提供了新的解决方案。
44
45==================================================
46错误处理示例
47==================================================
48INFO:httpx:HTTP Request: POST http://127.0.0.1:11434/api/chat "HTTP/1.1 200 OK"
49成功：您好！我注意到您提到了"测试问题"，但我没有看到具体需要测试的内容。为了更好地帮助您，能否请您：
50
511. **说明具体需要测试什么？**
52   - 是技术问题、逻辑问题、还是其他类型的测试？
53   - 请提供详细的问题描述
54
552. **告知测试的背景**
56   - 这是关于什么领域的测试？
57   - 有什么特殊要求吗？
58
593. **提供相关材料**
60   - 如果有代码、数据或其他资料，请一并提供
61
62请补充这些信息，我就能为您提供更准确和有用的测试帮助了。期待您的详细说明！
63
64==================================================
65Token 计数示例
66==================================================
67⚠ Ollama 本地模型不支持 OpenAI Token 计数，此示例仅供 OpenAI API 使用
68

生产环境最佳实践

1sequenceDiagram
2    participant 用户
3    participant 日志系统
4    participant 限流器
5    participant LLM客户端
6    participant Ollama模型
7
8    用户->>日志系统: 发送请求 "问题 X"
9    日志系统->>日志系统: 记录收到的请求
10    日志系统->>限流器: 检查请求是否允许
11    alt 请求被允许
12        限流器-->>日志系统: 允许
13        日志系统->>LLM客户端: 调用 safe_generate()
14        LLM客户端->>LLM客户端: validate_input()
15        alt 输入有效
16            LLM客户端->>LLM客户端: retry_on_failure
17            LLM客户端->>Ollama模型: invoke(prompt)
18            Ollama模型-->>LLM客户端: 返回生成内容
19            LLM客户端-->>日志系统: 响应生成成功
20            日志系统-->>用户: 返回响应内容
21        else 输入无效
22            LLM客户端-->>日志系统: 输入验证失败
23            日志系统-->>用户: 返回错误信息
24        end
25    else 请求被限制
26        限流器-->>日志系统: 超过速率限制
27        日志系统-->>用户: 返回限制提示
28    end
29
30

创建文件 production_best_practices.py：

1"""
2生产环境部署最佳实践（LangChain 1.0 + Ollama）
3"""
4
5from langchain_ollama import ChatOllama
6from dotenv import load_dotenv
7import os
8import time
9from functools import wraps
10from collections import deque
11import logging
12
13load_dotenv()
14
15# =========================
16# 重试装饰器
17# =========================
18def retry_on_failure(max_retries=3, delay=1):
19    """自动重试失败的请求"""
20    def decorator(func):
21        @wraps(func)
22        def wrapper(*args, **kwargs):
23            for attempt in range(max_retries):
24                try:
25                    return func(*args, **kwargs)
26                except Exception as e:
27                    if attempt < max_retries - 1:
28                        print(f"尝试 {attempt + 1} 失败: {e}")
29                        print(f"等待 {delay} 秒后重试...")
30                        time.sleep(delay * (attempt + 1))
31                    else:
32                        print(f"所有 {max_retries} 次尝试均失败")
33                        raise
34        return wrapper
35    return decorator
36
37# =========================
38# 生产环境 LLM 客户端
39# =========================
40class ProductionLLMClient:
41    """生产环境 LLM 客户端封装（Ollama）"""
42    def __init__(self):
43        # 从环境变量读取配置
44        self.model = os.getenv("LLM_MODEL", "Qwen2.5:7b")
45        self.temperature = float(os.getenv("LLM_TEMPERATURE", "0.7"))
46        self.max_tokens = int(os.getenv("LLM_MAX_TOKENS", "500"))
47
48        # 初始化 Ollama LLM
49        self.llm = ChatOllama(
50            model=self.model,
51            temperature=self.temperature,
52            max_tokens=self.max_tokens,
53            streaming=False
54        )
55
56        print(f"✓ LLM 客户端初始化完成")
57        print(f"  模型: {self.model}")
58        print(f"  温度: {self.temperature}")
59        print(f"  最大 Token: {self.max_tokens}")
60
61    @retry_on_failure(max_retries=3, delay=2)
62    def generate(self, prompt):
63        """生成响应（带重试机制）"""
64        try:
65            response = self.llm.invoke(prompt)
66            return response.content
67        except Exception as e:
68            print(f"生成失败: {e}")
69            raise
70
71    def validate_input(self, text, max_length=1000):
72        """验证输入合法性"""
73        if not text or not isinstance(text, str):
74            return False
75        if len(text) > max_length:
76            return False
77        return True
78
79    def safe_generate(self, user_input):
80        """安全生成响应"""
81        if not self.validate_input(user_input):
82            return {"success": False, "error": "输入无效或过长"}
83        try:
84            response = self.generate(user_input)
85            return {"success": True, "response": response}
86        except Exception as e:
87            return {"success": False, "error": str(e)}
88
89# =========================
90# 日志示例
91# =========================
92def logging_example():
93    logging.basicConfig(
94        level=logging.INFO,
95        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
96        handlers=[logging.FileHandler('langchain_app.log'), logging.StreamHandler()]
97    )
98    logger = logging.getLogger(__name__)
99
100    print("=" * 50)
101    print("日志记录示例")
102    print("=" * 50)
103
104    client = ProductionLLMClient()
105    user_input = "什么是机器学习？"
106    logger.info(f"收到用户请求: {user_input}")
107
108    result = client.safe_generate(user_input)
109    if result["success"]:
110        logger.info("响应生成成功")
111        print(f"\n响应: {result['response']}")
112    else:
113        logger.error(f"响应生成失败: {result['error']}")
114        print(f"\n错误: {result['error']}")
115
116# =========================
117# 速率限制示例
118# =========================
119def rate_limiting_example():
120    class RateLimiter:
121        """简单速率限制器"""
122        def __init__(self, max_calls, time_window):
123            self.max_calls = max_calls
124            self.time_window = time_window
125            self.calls = deque()
126
127        def allow_request(self):
128            now = time.time()
129            while self.calls and self.calls[0] < now - self.time_window:
130                self.calls.popleft()
131            if len(self.calls) < self.max_calls:
132                self.calls.append(now)
133                return True
134            return False
135
136    print("\n" + "=" * 50)
137    print("速率限制示例")
138    print("=" * 50)
139
140    limiter = RateLimiter(max_calls=5, time_window=60)
141    client = ProductionLLMClient()
142
143    for i in range(7):
144        if limiter.allow_request():
145            print(f"\n请求 {i+1}: 允许")
146            result = client.safe_generate(f"问题 {i+1}")
147            print(f"响应: {result.get('response', result.get('error'))[:50]}...")
148        else:
149            print(f"\n请求 {i+1}: 被限制（超过速率限制）")
150
151# =========================
152# 主函数
153# =========================
154if __name__ == "__main__":
155    logging_example()
156    rate_limiting_example()
157
158