LangChain AI开发框架配置指南

框架简介

LangChain 是一个强大的框架，用于开发基于语言模型的应用程序。通过老张API，您可以在 LangChain 中使用各种主流 AI 模型。

快速开始

安装

pip install langchain langchain-openai

基础配置

import os
from langchain_openai import ChatOpenAI

# 设置环境变量
os.environ["OPENAI_API_KEY"] = "您的老张API密钥"
os.environ["OPENAI_BASE_URL"] = "https://api2.laozhang.ai/v1"

# 初始化模型
llm = ChatOpenAI(
    model="gpt-3.5-turbo",
    temperature=0.7
)

核心功能

基础对话

from langchain.schema import HumanMessage, SystemMessage

messages = [
    SystemMessage(content="你是一个有帮助的助手"),
    HumanMessage(content="介绍一下 Python 的主要特点")
]

response = llm.invoke(messages)
print(response.content)

对话链

from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain

# 创建带记忆的对话链
memory = ConversationBufferMemory()
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

# 多轮对话
conversation.predict(input="我想学习机器学习")
conversation.predict(input="推荐一些入门资源")

文档问答系统

from langchain.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS
from langchain.chains import RetrievalQA

# 配置嵌入模型
embeddings = OpenAIEmbeddings(
    api_key="您的老张API密钥",
    base_url="https://api2.laozhang.ai/v1"
)

# 加载文档
loader = TextLoader("document.txt")
documents = loader.load()

# 分割文本
text_splitter = CharacterTextSplitter(
    chunk_size=1000, 
    chunk_overlap=0
)
texts = text_splitter.split_documents(documents)

# 创建向量存储
vectorstore = FAISS.from_documents(texts, embeddings)

# 创建问答链
qa = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever()
)

# 提问
result = qa.run("文档中的关键概念是什么？")

模型切换

使用不同模型

# GPT-4
gpt4 = ChatOpenAI(
    model="gpt-4",
    api_key="您的老张API密钥",
    base_url="https://api2.laozhang.ai/v1"
)

# Claude 3
claude = ChatOpenAI(
    model="claude-3-opus-20240229",
    api_key="您的老张API密钥", 
    base_url="https://api2.laozhang.ai/v1"
)

# 比较不同模型的回答
question = "解释量子计算的基本原理"
gpt4_answer = gpt4.invoke([HumanMessage(content=question)])
claude_answer = claude.invoke([HumanMessage(content=question)])

高级应用

Agent 系统

from langchain.agents import create_openai_functions_agent, AgentExecutor
from langchain.tools import Tool
from langchain import hub

# 定义工具
def get_weather(location: str) -> str:
    """获取天气信息"""
    return f"{location}的天气是晴天，温度25°C"

weather_tool = Tool(
    name="Weather",
    func=get_weather,
    description="获取指定地点的天气信息"
)

# 创建 Agent
prompt = hub.pull("hwchase17/openai-functions-agent")
agent = create_openai_functions_agent(llm, [weather_tool], prompt)
agent_executor = AgentExecutor(agent=agent, tools=[weather_tool])

# 使用 Agent
result = agent_executor.invoke({"input": "北京今天天气怎么样？"})

批量处理

# 批量生成响应
prompts = [
    "解释人工智能",
    "什么是机器学习", 
    "深度学习的应用"
]

responses = llm.batch([
    HumanMessage(content=p) for p in prompts
])

for response in responses:
    print(response.content)
    print("-" * 50)

流式输出

from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

# 配置流式输出
streaming_llm = ChatOpenAI(
    model="gpt-3.5-turbo",
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()]
)

# 流式生成
streaming_llm.invoke("写一首关于春天的诗")

错误处理

from langchain.callbacks import get_openai_callback

try:
    with get_openai_callback() as cb:
        response = llm.invoke("你好")
        print(f"使用的 Token 数: {cb.total_tokens}")
        print(f"API 调用成本: ${cb.total_cost}")
except Exception as e:
    print(f"发生错误: {e}")

最佳实践

1. 模型选择策略

任务类型	推荐模型	原因
简单对话	gpt-3.5-turbo	快速经济
复杂推理	gpt-4	准确度高
长文本	claude-3-opus	上下文长
创意写作	claude-3-sonnet	创造力强

2. 成本优化

class CostOptimizedLLM:
    def __init__(self):
        self.light_model = ChatOpenAI(model="gpt-3.5-turbo")
        self.premium_model = ChatOpenAI(model="gpt-4")
    
    def smart_invoke(self, message, complexity="low"):
        model = self.premium_model if complexity == "high" else self.light_model
        return model.invoke(message)

3. 缓存策略

from langchain.cache import InMemoryCache
from langchain.globals import set_llm_cache

# 启用缓存
set_llm_cache(InMemoryCache())

# 相同请求会使用缓存
response1 = llm.invoke("什么是人工智能？")
response2 = llm.invoke("什么是人工智能？")  # 使用缓存

4. 异步处理

import asyncio
from langchain_openai import AsyncChatOpenAI

async def async_chat():
    async_llm = AsyncChatOpenAI(
        model="gpt-3.5-turbo",
        api_key="您的老张API密钥",
        base_url="https://api2.laozhang.ai/v1"
    )
    
    response = await async_llm.ainvoke("异步生成的内容")
    return response.content

# 运行异步函数
result = asyncio.run(async_chat())

复杂应用示例

多模态 RAG 系统

class MultiModalRAG:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o")  # 支持图像
        self.embeddings = OpenAIEmbeddings()
        self.vectorstore = None
    
    def add_documents(self, documents):
        """添加文档到知识库"""
        texts = self.text_splitter.split_documents(documents)
        self.vectorstore = FAISS.from_documents(texts, self.embeddings)
    
    def query_with_image(self, question, image_url):
        """支持图像的查询"""
        context = self.vectorstore.similarity_search(question, k=3)
        
        messages = [
            {"role": "system", "content": "基于提供的文档回答问题"},
            {"role": "user", "content": [
                {"type": "text", "text": f"问题: {question}\n\n上下文: {context}"},
                {"type": "image_url", "image_url": {"url": image_url}}
            ]}
        ]
        
        return self.llm.invoke(messages)

智能工作流

from langchain.schema.runnable import RunnableLambda, RunnableSequence

def classify_intent(query):
    """意图分类"""
    classifier = ChatOpenAI(model="gpt-3.5-turbo")
    result = classifier.invoke(f"将以下查询分类为：问答、创作、分析、其他\n\n{query}")
    return result.content.strip()

def route_to_specialist(intent_and_query):
    """路由到专门的处理器"""
    intent, query = intent_and_query
    
    if "问答" in intent:
        model = ChatOpenAI(model="gpt-3.5-turbo")
    elif "创作" in intent:
        model = ChatOpenAI(model="claude-3-sonnet")
    elif "分析" in intent:
        model = ChatOpenAI(model="gpt-4")
    else:
        model = ChatOpenAI(model="gpt-3.5-turbo")
    
    return model.invoke(query)

# 创建工作流
workflow = RunnableSequence(
    RunnableLambda(lambda x: (classify_intent(x), x)),
    RunnableLambda(route_to_specialist)
)

# 使用工作流
result = workflow.invoke("帮我分析这个季度的销售数据")

性能监控

import time
from functools import wraps

def monitor_llm_calls(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.time()
        try:
            result = func(*args, **kwargs)
            success = True
        except Exception as e:
            result = None
            success = False
            print(f"LLM调用失败: {e}")
        
        end_time = time.time()
        duration = end_time - start_time
        
        print(f"LLM调用 - 成功: {success}, 耗时: {duration:.2f}s")
        return result
    
    return wrapper

@monitor_llm_calls
def safe_llm_call(llm, message):
    return llm.invoke(message)

部署建议

1. 生产环境配置

import os
from langchain_openai import ChatOpenAI

class ProductionLLM:
    def __init__(self):
        self.llm = ChatOpenAI(
            model=os.getenv("LLM_MODEL", "gpt-3.5-turbo"),
            temperature=float(os.getenv("LLM_TEMPERATURE", "0.7")),
            max_tokens=int(os.getenv("LLM_MAX_TOKENS", "1000")),
            request_timeout=int(os.getenv("LLM_REQUEST_TIMEOUT", "60"))
        )
    
    def chat(self, message):
        try:
            return self.llm.invoke(message)
        except Exception as e:
            # 记录错误日志
            print(f"LLM错误: {e}")
            return "抱歉，服务暂时不可用"

2. 容错机制

def retry_llm_call(max_retries=3, delay=1):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise e
                    time.sleep(delay * (2 ** attempt))  # 指数退避
            return wrapper
    return decorator

@retry_llm_call(max_retries=3)
def robust_llm_call(llm, message):
    return llm.invoke(message)

需要更多帮助？请查看使用场景总览。

编程开发

概览

对话型 AI

技术工程

翻译场景

LangChain AI开发框架配置指南

框架简介

快速开始

安装

基础配置

核心功能

基础对话

对话链

文档问答系统

模型切换

使用不同模型

高级应用

Agent 系统

批量处理

流式输出

错误处理

最佳实践

1. 模型选择策略

2. 成本优化

3. 缓存策略

4. 异步处理

复杂应用示例

多模态 RAG 系统

智能工作流

性能监控

部署建议

1. 生产环境配置

2. 容错机制

​框架简介

​快速开始

​安装

​基础配置

​核心功能

​基础对话

​对话链

​文档问答系统

​模型切换

​使用不同模型

​高级应用

​Agent 系统

​批量处理

​流式输出

​错误处理

​最佳实践

​1. 模型选择策略

​2. 成本优化

​3. 缓存策略

​4. 异步处理

​复杂应用示例

​多模态 RAG 系统

​智能工作流

​性能监控

​部署建议

​1. 生产环境配置

​2. 容错机制

框架简介

快速开始

安装

基础配置

核心功能

基础对话

对话链

文档问答系统

模型切换

使用不同模型

高级应用

Agent 系统

批量处理

流式输出

错误处理

最佳实践

1. 模型选择策略

2. 成本优化

3. 缓存策略

4. 异步处理

复杂应用示例

多模态 RAG 系统

智能工作流

性能监控

部署建议

1. 生产环境配置

2. 容错机制