LangChain LCEL 框架¶

import langchain
import langchain_core
import langgraph

print(f"langchain: {langchain.__version__}")
print(f"langchain-core: {langchain_core.__version__}")
print(f"langgraph: {langgraph.__version__}")

import os
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

# 设置API密钥
os.environ['OPENAI_API_KEY'] = 'your-api-key'

# 初始化模型
llm = ChatOpenAI(
    model_name="gpt-5.4",
    temperature=0.7
)

# 初始化嵌入模型
embeddings = OpenAIEmbeddings(
    openai_api_key='your-api-key'
)

from langchain_openai import ChatOpenAI, OpenAI
from langchain_anthropic import ChatAnthropic
from langchain_google_genai import ChatGoogleGenerativeAI

# OpenAI Chat模型
chat = ChatOpenAI(
    model_name="gpt-5.4",
    temperature=0.7,
    max_tokens=1000
)

# OpenAI 经济型模型
llm = ChatOpenAI(
    model="gpt-5-mini",
    temperature=0.7
)

# Anthropic模型
# 请替换为你当前账户与地区实际可用的 Claude 模型 ID
claude = ChatAnthropic(
    model="your_available_claude_model",
    temperature=0.7
)

# Google模型
# 请替换为当前可用的 Gemini 模型 ID
gemini = ChatGoogleGenerativeAI(
    model="your_available_gemini_model",
    temperature=0.7
)

# 使用模型
response = chat.invoke("你好!")
print(response.content)

from langchain_openai import OpenAIEmbeddings
from langchain_huggingface import HuggingFaceEmbeddings

# OpenAI嵌入
openai_embeddings = OpenAIEmbeddings(
    openai_api_key='your-api-key'
)

# HuggingFace嵌入
hf_embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-MiniLM-L6-v2"
)

# 生成嵌入
text = "这是一段示例文本"
embedding = openai_embeddings.embed_query(text)
print(f"嵌入维度: {len(embedding)}")

from langchain_core.prompts import PromptTemplate, ChatPromptTemplate, MessagesPlaceholder

# 基础Prompt模板
template = """
你是一个专业的{role}。
请回答以下问题: {question}
"""

prompt = PromptTemplate(
    template=template,
    input_variables=["role", "question"]
)

# 使用模板
formatted_prompt = prompt.format(
    role="技术顾问",
    question="什么是机器学习?"
)
print(formatted_prompt)

# ChatPrompt模板
chat_template = ChatPromptTemplate.from_messages([
    ("system", "你是一个{role}。"),
    ("user", "{question}")
])

chat_prompt = chat_template.format_messages(
    role="技术顾问",
    question="什么是机器学习?"
)
print(chat_prompt)

# 使用MessagesPlaceholder
messages_template = ChatPromptTemplate.from_messages([
    ("system", "你是一个助手。"),
    MessagesPlaceholder(variable_name="history"),
    ("user", "{input}")
])

from langchain_core.prompts import FewShotPromptTemplate
from langchain_core.prompts import PromptTemplate

# 定义示例
examples = [
    {
        "question": "苹果是什么?",
        "answer": "苹果是一种水果,也可以指苹果公司。"
    },
    {
        "question": "香蕉是什么?",
        "answer": "香蕉是一种热带水果,富含钾元素。"
    }
]

# 定义示例模板
example_prompt = PromptTemplate(
    input_variables=["question", "answer"],
    template="问题: {question}\n答案: {answer}"
)

# 创建Few-shot Prompt
few_shot_prompt = FewShotPromptTemplate(
    examples=examples,
    example_prompt=example_prompt,
    prefix="以下是一些示例:",
    suffix="\n问题: {input}\n答案:",
    input_variables=["input"],
    example_separator="\n\n"
)

# 使用
formatted = few_shot_prompt.format(input="橙子是什么?")
print(formatted)

# 使用LCEL（LangChain Expression Language）创建链
chain = prompt | chat

# 运行链
result = chain.invoke({
    "role": "技术顾问",
    "question": "什么是机器学习?"
})
print(result)

from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

# 第一个链: 生成介绍
prompt1 = PromptTemplate.from_template("简要介绍{topic}。")
chain1 = prompt1 | chat | StrOutputParser()

# 第二个链: 提取要点
prompt2 = PromptTemplate.from_template(
    "基于以下介绍,列出3个要点:\n{introduction}"
)
chain2 = prompt2 | chat | StrOutputParser()

# LCEL组合: 管道串联
overall_chain = (
    {"introduction": chain1, "topic": RunnablePassthrough()}
    | RunnablePassthrough.assign(points=chain2)
)

# 运行
result = overall_chain.invoke("机器学习")
print(result["introduction"])
print(result["points"])

from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableBranch, RunnableLambda

# 路由分类器
router_prompt = PromptTemplate.from_template(
    """判断以下问题的类型,只输出 tech 或 general:
问题: {input}
类型:"""
)
router_chain = router_prompt | chat | StrOutputParser()

# 目标链
tech_chain = PromptTemplate.from_template("作为技术专家回答: {input}") | chat | StrOutputParser()
general_chain = PromptTemplate.from_template("作为助手回答: {input}") | chat | StrOutputParser()

# LCEL路由
def route(info):
    route_type = info["route"].strip().lower()  # 链式调用：strip去除空白
    if "tech" in route_type:
        return tech_chain.invoke({"input": info["input"]})
    return general_chain.invoke({"input": info["input"]})

full_chain = (
    {"route": router_chain, "input": lambda x: x["input"]}  # lambda匿名函数
    | RunnableLambda(route)
)

# 使用
result = full_chain.invoke({"input": "什么是Transformer?"})
print(result)

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# 创建带记忆的Prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的AI助手。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])

chain = prompt | chat

# 记忆存储（按session_id隔离）
message_store = {}
def get_session_history(session_id: str):
    if session_id not in message_store:
        message_store[session_id] = InMemoryChatMessageHistory()
    return message_store[session_id]

# 带记忆的链
conversation = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history"
)

config = {"configurable": {"session_id": "user-001"}}

# 对话
response1 = conversation.invoke({"input": "你好!"}, config=config)
print(f"AI: {response1.content}")

response2 = conversation.invoke({"input": "我叫小明"}, config=config)
print(f"AI: {response2.content}")

response3 = conversation.invoke({"input": "我叫什么名字?"}, config=config)
print(f"AI: {response3.content}")

# 查看记忆
print("\n对话历史:")
for msg in message_store["user-001"].messages:
    print(f"{msg.type}: {msg.content}")

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

class SummarizingChatHistory:
    """当对话超过阈值时自动摘要压缩"""

    def __init__(self, llm, max_messages=10):
        self.llm = llm
        self.max_messages = max_messages
        self.messages = []
        self.summary = ""

    def add_message(self, message):
        self.messages.append(message)
        if len(self.messages) > self.max_messages:
            self._summarize()

    def _summarize(self):
        """LLM生成摘要，替代旧消息"""
        history_text = "\n".join(
            f"{m.type}: {m.content}" for m in self.messages[:-2]
        )
        summary_prompt = f"请用1-2句话概括以下对话:\n{history_text}"
        result = self.llm.invoke(summary_prompt)
        self.summary = result.content
        self.messages = self.messages[-2:]  # 只保留最近2轮

    def get_messages(self):
        msgs = []
        if self.summary:
            msgs.append(SystemMessage(content=f"之前的对话摘要: {self.summary}"))
        msgs.extend(self.messages)
        return msgs

# 使用
history = SummarizingChatHistory(llm=chat, max_messages=6)
history.add_message(HumanMessage(content="介绍一下机器学习"))
result = chat.invoke(history.get_messages())
print(f"AI: {result.content}")
history.add_message(AIMessage(content=result.content))

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的助手。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])

chain = prompt | chat

# 带窗口限制的记忆
class WindowedChatHistory(InMemoryChatMessageHistory):
    """K窗口记忆：只保留最近k轮对话"""
    def __init__(self, k=2):
        super().__init__()  # super()调用父类方法
        self.k = k

    def add_message(self, message):
        super().add_message(message)
        # 每轮对话=2条消息(human+AI)，保留k轮
        if len(self.messages) > self.k * 2:
            self.messages = self.messages[-(self.k * 2):]

store = {}
def get_windowed_history(session_id):
    if session_id not in store:
        store[session_id] = WindowedChatHistory(k=2)
    return store[session_id]

conversation = RunnableWithMessageHistory(
    chain, get_windowed_history,
    input_messages_key="input",
    history_messages_key="history"
)

config = {"configurable": {"session_id": "user-002"}}

# 多轮对话
for i in range(5):
    user_input = input("用户: ")
    response = conversation.invoke({"input": user_input}, config=config)
    print(f"AI: {response.content}")

from langchain_community.tools import TavilySearchResults
from langchain_community.tools import WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper
from langgraph.prebuilt import create_react_agent

# 创建工具
search_tool = TavilySearchResults(max_results=3)
wiki_tool = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())

tools = [search_tool, wiki_tool]

# 创建ReAct Agent（LangGraph推荐方式）
agent = create_react_agent(chat, tools)

# 运行
result = agent.invoke({"messages": [{"role": "user", "content": "最新的AI技术进展是什么?"}]})
print(result["messages"][-1].content)  # [-1]负索引取最后一个元素

from langchain_core.tools import BaseTool
from typing import Type
from pydantic import BaseModel, Field

class CalculatorInput(BaseModel):  # Pydantic BaseModel：自动数据验证和序列化
    """计算器输入"""
    expression: str = Field(description="数学表达式")

class CalculatorTool(BaseTool):
    """计算器工具"""

    name = "calculator"
    description = "用于计算数学表达式"
    args_schema: Type[BaseModel] = CalculatorInput

    def _run(self, expression: str) -> str:
        """运行工具（使用AST安全解析）"""
        import ast
        try:  # try/except捕获异常，防止程序崩溃
            tree = ast.parse(expression, mode='eval')
            for node in ast.walk(tree):
                if not isinstance(node, (ast.Expression, ast.BinOp, ast.UnaryOp,  # isinstance检查类型
                                        ast.Constant, ast.Add, ast.Sub, ast.Mult,
                                        ast.Div, ast.Mod, ast.Pow, ast.USub)):
                    return f"计算错误: 不允许的表达式"
            result = eval(compile(tree, '<expr>', 'eval'), {"__builtins__": {}}, {})
            return str(result)
        except Exception as e:
            return f"计算错误: {str(e)}"

    async def _arun(self, expression: str) -> str:  # async def定义协程函数
        """异步运行"""
        return self._run(expression)

# 使用工具
calculator = CalculatorTool()
result = calculator._run("2 + 3 * 4")
print(f"计算结果: {result}")

# 添加到LangGraph Agent
from langgraph.prebuilt import create_react_agent

tools = [calculator]
agent = create_react_agent(chat, tools)

result = agent.invoke({"messages": [{"role": "user", "content": "计算 2 + 3 * 4"}]})
print(result["messages"][-1].content)

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

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

# 分割文档
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
splits = text_splitter.split_documents(documents)

# 创建向量存储
vectorstore = Chroma.from_documents(
    documents=splits,
    embedding=embeddings
)

# 创建检索器
retriever = vectorstore.as_retriever(
    search_kwargs={"k": 3}
)

# LCEL构建RAG链（替代已弃用的RetrievalQA）
prompt = ChatPromptTemplate.from_template(
    """基于以下上下文回答问题。如果上下文中没有相关信息，请说“我无法根据提供的信息回答”。

上下文: {context}
问题: {question}
答案:"""
)

def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | chat
    | StrOutputParser()
)

# 查询
query = "文档的主要内容是什么?"
answer = rag_chain.invoke(query)
print(f"答案: {answer}")

# 如需返回来源文档，可用RunnableParallel
from langchain_core.runnables import RunnableParallel

rag_with_sources = RunnableParallel(
    answer=rag_chain,
    sources=retriever
)
result = rag_with_sources.invoke(query)
print(f"答案: {result['answer']}")
print(f"来源: {[doc.page_content[:50] for doc in result['sources']]}")

from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

# 定义工具（注意：用@tool装饰器替代Tool类）
@tool
def search(query: str) -> str:
    """搜索工具，用于搜索信息"""
    # 这里简化处理
    return f"搜索结果: {query}"

@tool
def calculator(expression: str) -> str:
    """计算器工具，用于数学计算（使用AST安全解析）"""
    import ast
    try:
        tree = ast.parse(expression, mode='eval')
        for node in ast.walk(tree):
            if not isinstance(node, (ast.Expression, ast.BinOp, ast.UnaryOp,
                                    ast.Constant, ast.Add, ast.Sub, ast.Mult,
                                    ast.Div, ast.Mod, ast.Pow, ast.USub)):
                return f"计算错误: 不允许的表达式"
        result = eval(compile(tree, '<expr>', 'eval'), {"__builtins__": {}}, {})
        return str(result)
    except Exception as e:
        return f"计算错误: {str(e)}"

# 创建LangGraph ReAct Agent
tools = [search, calculator]
agent = create_react_agent(chat, tools)

# 运行Agent
result = agent.invoke(
    {"messages": [{"role": "user", "content": "搜索AI技术进展并计算 2 + 3 * 4"}]}
)

# 提取最终回复
print(result["messages"][-1].content)

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

# LCEL对话链
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的AI助手。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])

chain = prompt | chat
store = {}
def get_history(sid):
    if sid not in store:
        store[sid] = InMemoryChatMessageHistory()
    return store[sid]

conversation = RunnableWithMessageHistory(
    chain, get_history,
    input_messages_key="input",
    history_messages_key="history"
)

# 多模态输入示例（OpenAI 多模态模型支持图片 URL / base64）
from langchain_core.messages import HumanMessage

multimodal_msg = HumanMessage(content=[
    {"type": "text", "text": "请分析这张图片的内容"},
    {"type": "image_url", "image_url": {
        "url": "https://example.com/screenshot.png"
    }}
])

response = chat.invoke([multimodal_msg])
print(f"AI: {response.content}")

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# LCEL流式链
prompt = ChatPromptTemplate.from_template("写一首关于{topic}的短诗")
chain = prompt | chat | StrOutputParser()

# 流式输出
print("=== 流式输出 ===")
for chunk in chain.stream({"topic": "人工智能"}):
    print(chunk, end="", flush=True)
print()  # 换行

# 异步流式
import asyncio

async def async_stream():
    print("=== 异步流式输出 ===")
    async for chunk in chain.astream({"topic": "春天"}):
        print(chunk, end="", flush=True)
    print()

# asyncio.run(async_stream())  # 在async环境中调用

# 异步调用
async def async_invoke():
    result = await chain.ainvoke({"topic": "未来"})
    print(f"异步结果: {result}")

# 批量异步
async def async_batch():
    topics = ["AI", "区块链", "量子计算"]
    results = await chain.abatch([{"topic": t} for t in topics])
    for topic, result in zip(topics, results):
        print(f"{topic}: {result[:50]}...")

# asyncio.run(async_batch())

from langchain_core.runnables import RunnableParallel, RunnableLambda

# 定义多个处理分支
def analyze_sentiment(text: str) -> str:
    """情感分析"""
    positive_words = ["好", "棒", "优秀", "喜欢"]
    negative_words = ["差", "糟", "糟糕", "讨厌"]

    pos_count = sum(1 for w in positive_words if w in text)
    neg_count = sum(1 for w in negative_words if w in text)

    if pos_count > neg_count:
        return "正面"
    elif neg_count > pos_count:
        return "负面"
    return "中性"

def extract_keywords(text: str) -> list:
    """关键词提取（简化版）"""
    # 实际应用中可用jieba或LLM
    words = text.split()
    return [w for w in words if len(w) > 2][:5]

def count_words(text: str) -> int:
    """字数统计"""
    return len(text.replace(" ", ""))

# 并行处理链
parallel_chain = RunnableParallel(
    sentiment=RunnableLambda(analyze_sentiment),
    keywords=RunnableLambda(extract_keywords),
    word_count=RunnableLambda(count_words)
)

# 测试
text = "这个产品真的很棒，我非常喜欢它的设计"
result = parallel_chain.invoke(text)
print(f"情感: {result['sentiment']}")
print(f"关键词: {result['keywords']}")
print(f"字数: {result['word_count']}")

from langchain_core.runnables import RunnablePassthrough

# 保留原始输入同时添加处理结果
chain_with_context = (
    {
        "original": RunnablePassthrough(),  # 原样传递
        "uppercase": RunnableLambda(lambda x: x.upper()),
        "length": RunnableLambda(len)
    }
)

result = chain_with_context.invoke("hello world")
print(result)
# {'original': 'hello world', 'uppercase': 'HELLO WORLD', 'length': 11}

from langchain_core.runnables import RunnableRetry

# 带重试的链
retry_chain = (
    prompt
    | chat
    | StrOutputParser()
).with_retry(
    stop_after_attempt=3,
    wait_exponential_jitter=True,
    retry_if_exception_type=(Exception,)
)

# 带超时的链
timeout_chain = chain.with_timeout(30)  # 30秒超时

# 带回退的链
fallback_prompt = ChatPromptTemplate.from_template("简单回答: {question}")
fallback_chain = fallback_prompt | chat | StrOutputParser()

robust_chain = chain.with_fallback([fallback_chain])

from langchain_core.callbacks import BaseCallbackHandler

class MyCallbackHandler(BaseCallbackHandler):
    """自定义回调处理器"""

    def on_llm_start(self, serialized, prompts, **kwargs):
        print(f"[LLM开始] Prompts: {prompts[:50]}...")

    def on_llm_end(self, response, **kwargs):
        print(f"[LLM结束] Token使用: {response.llm_output}")

    def on_chain_start(self, serialized, inputs, **kwargs):
        print(f"[链开始] 输入: {inputs}")

    def on_chain_end(self, outputs, **kwargs):
        print(f"[链结束] 输出长度: {len(str(outputs))}")

# 使用回调
chain_with_callback = chain.with_config(
    callbacks=[MyCallbackHandler()]
)
result = chain_with_callback.invoke({"topic": "科技"})

# 优化的Prompt模板
optimized_template = PromptTemplate(
    template="""
你是一个专业的{role},拥有{experience}年经验。

任务: {task}

要求:
1. {requirement1}
2. {requirement2}
3. {requirement3}

输入: {input}

请按照以上要求完成任务。
""",
    input_variables=[
        "role", "experience", "task",
        "requirement1", "requirement2", "requirement3", "input"
    ]
)

from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain.globals import set_llm_cache
from langchain_community.cache import InMemoryCache

# 启用缓存
set_llm_cache(InMemoryCache())

# LCEL链（替代已弃用的LLMChain）
chain = optimized_template | chat | StrOutputParser()

# 运行链
result = chain.invoke({
    "role": "技术顾问",
    "experience": "10",
    "task": "解释机器学习",
    "requirement1": "使用简洁的语言",
    "requirement2": "提供实际例子",
    "requirement3": "避免过于专业的术语",
    "input": "什么是机器学习?"
})
print(result)

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# LCEL对话链 + 持久化记忆
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的助手。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])

chain = prompt | chat

# 使用文件存储记忆（也可换成Redis/PostgreSQL）
from langchain_community.chat_message_histories import FileChatMessageHistory

def get_file_history(session_id: str):
    return FileChatMessageHistory(file_path=f"chat_{session_id}.json")

conversation = RunnableWithMessageHistory(
    chain, get_file_history,
    input_messages_key="input",
    history_messages_key="history"
)

from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 创建Prompt模板
template = PromptTemplate.from_template("回答以下问题: {question}")

# LCEL链（替代已弃用的LLMChain）
chain = template | chat | StrOutputParser()

# 运行链
result = chain.invoke({"question": "什么是机器学习?"})
print(result)

from langchain_core.tools import BaseTool
from typing import Type
from pydantic import BaseModel, Field

class WeatherInput(BaseModel):
    """天气输入"""
    city: str = Field(description="城市名称")

class WeatherTool(BaseTool):
    """天气工具"""

    name = "weather"
    description = "用于查询天气信息"
    args_schema: Type[BaseModel] = WeatherInput

    def _run(self, city: str) -> str:
        """运行工具"""
        # 这里简化处理,实际应该调用天气API
        return f"{city}的天气: 晴, 温度: 25°C"

    async def _arun(self, city: str) -> str:
        """异步运行"""
        return self._run(city)