Langchain + LLM API 完整集成指南

# 安装 Langchain
pip install langchain langchain-openai chromadb tiktoken

# 基础配置示例
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.chains import LLMChain

# 初始化模型
llm = ChatOpenAI(
    base_url="https://api.n1n.ai/v1",
    api_key="your-api-key",
    model="gpt-4o",
    temperature=0.7
)

# 创建提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的{role}"),
    ("human", "{input}")
])

# 创建链
chain = LLMChain(llm=llm, prompt=prompt)

# 运行链
response = chain.run(role="翻译专家", input="Hello World")
print(response)

from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain_community.document_loaders import PyPDFLoader
import os

# 配置 API
os.environ["OPENAI_API_BASE"] = "https://api.n1n.ai/v1"
os.environ["OPENAI_API_KEY"] = "your-api-key"

# 1. 加载文档
loader = PyPDFLoader("document.pdf")
documents = loader.load()

# 2. 分割文档
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    length_function=len,
    separators=["\n\n", "\n", "。", "！", "？", "，", " ", ""]
)
texts = text_splitter.split_documents(documents)

# 3. 创建向量存储
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    openai_api_base="https://api.n1n.ai/v1"
)

vectorstore = Chroma.from_documents(
    documents=texts,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# 4. 创建检索链
llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0,
    openai_api_base="https://api.n1n.ai/v1"
)

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(
        search_type="similarity",
        search_kwargs={"k": 3}
    ),
    return_source_documents=True
)

# 5. 查询
query = "文档中提到的主要观点是什么？"
result = qa_chain({"query": query})

print("答案：", result["result"])
print("\n相关文档：")
for doc in result["source_documents"]:
    print(f"- {doc.page_content[:100]}...")

from langchain.agents import create_openai_functions_agent, AgentExecutor
from langchain.tools import Tool, StructuredTool
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage
import requests
from pydantic import BaseModel, Field

# 定义工具
class WeatherInput(BaseModel):
    city: str = Field(description="城市名称")

def get_weather(city: str) -> str:
    """获取天气信息"""
    # 实际应用中调用真实API
    return f"{city}的天气是晴天，温度25°C"

def search_web(query: str) -> str:
    """搜索网络信息"""
    return f"搜索结果：{query}的相关信息..."

# 创建工具列表
tools = [
    StructuredTool.from_function(
        func=get_weather,
        name="get_weather",
        description="获取指定城市的天气",
        args_schema=WeatherInput
    ),
    Tool(
        name="search_web",
        func=search_web,
        description="搜索网络获取信息"
    )
]

# 初始化模型
llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0,
    openai_api_base="https://api.n1n.ai/v1",
    openai_api_key="your-api-key"
)

# 创建提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个智能助手，可以使用以下工具：
    - get_weather: 获取天气信息
    - search_web: 搜索网络信息
    
    请根据用户的问题，选择合适的工具来回答。"""),
    MessagesPlaceholder(variable_name="chat_history", optional=True),
    ("human", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad")
])

# 创建代理
agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=3
)

# 执行代理
response = agent_executor.invoke({
    "input": "北京今天天气怎么样？明天适合出游吗？"
})

print(response["output"])

from langchain.memory import ConversationBufferMemory, ConversationSummaryMemory
from langchain_openai import ChatOpenAI
from langchain.chains import ConversationChain
from langchain.memory import ChatMessageHistory

# 1. 基础对话记忆
memory = ConversationBufferMemory()
llm = ChatOpenAI(
    model="gpt-4o",
    openai_api_base="https://api.n1n.ai/v1",
    openai_api_key="your-api-key"
)

conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

# 对话示例
print(conversation.run("我叫小明"))
print(conversation.run("我刚才说的名字是什么？"))

# 2. 摘要记忆（适合长对话）
summary_memory = ConversationSummaryMemory(
    llm=llm,
    max_token_limit=100
)

summary_chain = ConversationChain(
    llm=llm,
    memory=summary_memory,
    verbose=True
)

# 3. 向量记忆（检索相关历史）
from langchain.memory import VectorStoreRetrieverMemory
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

embeddings = OpenAIEmbeddings(
    openai_api_base="https://api.n1n.ai/v1"
)

vectorstore = Chroma(embedding_function=embeddings)
retriever = vectorstore.as_retriever(search_kwargs=dict(k=2))

vector_memory = VectorStoreRetrieverMemory(
    retriever=retriever,
    memory_key="history",
    input_key="input"
)

# 4. 持久化记忆
from langchain.memory import SQLiteEntityStore
from langchain.memory import ConversationEntityMemory

entity_store = SQLiteEntityStore(db_file="memory.db")
entity_memory = ConversationEntityMemory(
    llm=llm,
    entity_store=entity_store
)

from langchain_openai import ChatOpenAI
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
from langchain.prompts import ChatPromptTemplate
import asyncio

# 1. 同步流式输出
llm = ChatOpenAI(
    model="gpt-4o",
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()],
    openai_api_base="https://api.n1n.ai/v1",
    openai_api_key="your-api-key"
)

# 流式输出到控制台
response = llm.invoke("写一个关于AI的故事")

# 2. 异步流式处理
async def async_stream():
    llm = ChatOpenAI(
        model="gpt-4o",
        streaming=True,
        openai_api_base="https://api.n1n.ai/v1",
        openai_api_key="your-api-key"
    )
    
    async for chunk in llm.astream("讲一个笑话"):
        print(chunk.content, end="", flush=True)

# 3. 自定义流处理回调
from langchain.callbacks.base import BaseCallbackHandler
from typing import Any, Dict, List

class CustomStreamHandler(BaseCallbackHandler):
    def __init__(self):
        self.tokens = []
        
    def on_llm_new_token(self, token: str, **kwargs) -> None:
        """处理新的token"""
        self.tokens.append(token)
        # 可以发送到WebSocket、SSE等
        print(f"Token: {token}", end="")
        
    def on_llm_end(self, response: Any, **kwargs) -> None:
        """LLM结束时的处理"""
        full_response = "".join(self.tokens)
        print(f"\n完整响应长度: {len(full_response)}")

# 使用自定义处理器
custom_llm = ChatOpenAI(
    model="gpt-4o",
    streaming=True,
    callbacks=[CustomStreamHandler()],
    openai_api_base="https://api.n1n.ai/v1"
)

response = custom_llm.invoke("解释量子计算")

from langchain.cache import SQLiteCache, InMemoryCache
from langchain.globals import set_llm_cache
from langchain_openai import ChatOpenAI
import time

# 1. 设置缓存
set_llm_cache(SQLiteCache(database_path=".langchain.db"))
# 或使用内存缓存
# set_llm_cache(InMemoryCache())

llm = ChatOpenAI(
    model="gpt-3.5-turbo",  # 使用更便宜的模型
    openai_api_base="https://api.n1n.ai/v1",
    openai_api_key="your-api-key"
)

# 第一次调用（会缓存）
start = time.time()
response1 = llm.invoke("什么是机器学习？")
print(f"第一次调用耗时: {time.time() - start:.2f}秒")

# 第二次调用（从缓存读取）
start = time.time()
response2 = llm.invoke("什么是机器学习？")
print(f"第二次调用耗时: {time.time() - start:.2f}秒")

# 2. 批处理优化
from langchain.callbacks import get_openai_callback

prompts = [
    "解释Python",
    "解释JavaScript",
    "解释Rust"
]

# 使用回调跟踪成本
with get_openai_callback() as cb:
    # 批量处理
    responses = llm.batch(prompts, config={"max_concurrency": 3})
    print(f"总Token数: {cb.total_tokens}")
    print(f"预估成本: ${cb.total_cost:.4f}")

# 3. 提示压缩
from langchain.prompts import FewShotPromptTemplate
from langchain.prompts.example_selector import LengthBasedExampleSelector

examples = [
    {"input": "2+2", "output": "4"},
    {"input": "5*3", "output": "15"},
    {"input": "10/2", "output": "5"}
]

example_selector = LengthBasedExampleSelector(
    examples=examples,
    example_prompt=ChatPromptTemplate.from_template(
        "输入: {input}\n输出: {output}"
    ),
    max_length=50  # 限制示例长度
)

# 4. 模型降级策略
from langchain.llms import FallbackLLM

primary_llm = ChatOpenAI(
    model="gpt-4o",
    openai_api_base="https://api.n1n.ai/v1"
)

fallback_llm = ChatOpenAI(
    model="gpt-3.5-turbo",
    openai_api_base="https://api.n1n.ai/v1"
)

# 主模型失败时自动降级
try:
    response = primary_llm.invoke("复杂问题...")
except Exception:
    response = fallback_llm.invoke("复杂问题...")