Langchain

概念

LangChain 是一个用于构建由大语言模型（LLM）驱动的应用程序的开源框架。它覆盖了 LLM 应用全生命周期：

• 开发:使用 LangChain 的开源构建模块、组件 和 第三方集成构建您的应用程序。 使用 LangGraph 构建具有一流流式处理和人机协作支持的有状态代理。
• 生产化：使用 LangSmith 检查、监控和评估您的链，以便您可以持续优化并自信地部署
• 部署：将您的 LangGraph 应用程序转变为生产就绪的 API和助手，使用 LangGraph Cloud。

💡 实际项目中，通常只使用其开发能力，生产化和部署按需选用。

Langchain框架由以下开源库组成：

库名	作用
langchain-core	基础抽象层，定义 Runnable、LCEL 等核心接口
langchain-community	社区维护的第三方工具集成（如向量数据库、文档加载器）
langchain	高层应用组件：链（Chains）、代理（Agents）、检索（Retrieval）等
langchain-openai / langchain-anthropic 等	轻量级 LLM 集成包（仅依赖 langchain-core）
LangGraph	构建有状态、多步骤、支持人机协作的智能体（Agent），以图结构建模流程
LangServe	将 LangChain 链部署为 RESTful API
LangSmith	开发者平台，用于调试、测试、监控 LLM 应用

LCEL

LCEL的全称是LangChain Expression Language。其实他的用处就是使用“|”运算符 链接LangChain应用的各个组件。

特点：可组合、可并行、可流式、可回退、可监听，且无需修改代码即可用于生产环境。

LCEL的各种语法

开发时需要安装的库

langchain
langchain-openai
langchain-community
LangGraph

基础调用（非LCEL）

from langchain_core.messages import SystemMessage, HumanMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="GLM-4.5-Flash",
    temperature=1,
    openai_api_key='',
    openai_api_base=''
)
prompt = [
        SystemMessage('请把下面的语句翻译为日语'), # 内部大括号表示定义参数
		HumanMessage('今天天气怎么样？')
	]
parser = StrOutputParser() # 解析相应对象为字符串

# 第一种写法
res = llm.invoke(prompt)
parser.invoke(res)
print(res)
print(parser)

使用 LCEL 调用大模型（常用）

# 提示词模板
prompt = ChatPromptTemplate.from_messages(
	[
        ('system', '请把下面的语句翻译为{language}'), # 内部大括号表示定义参数
        ('user', '{user_text}')
	]
)
parser = StrOutputParser() # 解析相应对象为字符串

# LCEL
chain = prompt | llm | parser
res = chain.invoke({'language': '日文', 'user_text': '你吃饭了吗？'})

print(res)

1. Runnable节点

所有 LCEL 组件都继承自 Runnable，可通过 .invoke() 调用。

from langchain_core.runnables import RunnableLambda

def add_ten(x: int) -> int:
    return x + 10

node = RunnableLambda(add_ten)
print(node.invoke(5))  # 输出: 15

2. 多种执行模式

• 批量调用（Butch）

results = node.batch([1, 2, 3])  # [11, 12, 13]

• 流式输出

def word_stream(prompt: str):
    for word in prompt.split():
        yield word

stream_node = RunnableLambda(word_stream)
for chunk in stream_node.stream("Hello world from LCEL"):
    print(chunk)  # 逐词输出

3. 组合与并行

• 链式执行（Sequential）

double = RunnableLambda(lambda x: x * 2)
add_five = RunnableLambda(lambda x: x + 5)

# 节点组合为链，使用“|”
chain = double | add_five
print(chain.invoke(3))  # (3*2)+5 = 11

• 并行执行（Parallel）

from langchain_core.runnables import RunnableParallel

# 并行
parallel_chain = RunnableParallel(
    result1=double,
    result2=add_five
)
print(parallel_chain.invoke(4))  # {'result1': 8, 'result2': 9}

# 打印可视化流程图（需安装 graphviz）
parallel_chain.get_graph().print_ascii()

4. 中间数据处理：RunnablePassthrough

用于在链中保留、扩展或筛选输入数据。

from langchain_core.runnables import RunnablePassthrough

# 步骤1：包装输入为字典
wrap = RunnableLambda(lambda x: {"value": x})

# 步骤2：基于字典计算新字段
compute = RunnableLambda(lambda d: d["value"] * 2)

# 组合：保留原数据 + 添加新字段 + 仅输出新字段
chain = (
    wrap 
    | RunnablePassthrough.assign(doubled=compute) 
    | RunnablePassthrough.pick(["doubled"])
)

print(chain.invoke(5))  # {'doubled': 10}

常用方法：

• .assign(key=runnable)：添加新键
• .pick(keys)：筛选指定键
• RunnablePassthrough()：透传输入不变

5. 容错机制：后备选项（Fallbacks）

当主组件失败时，自动切换到备用方案。

def risky_func(x):
    if isinstance(x, str):
        raise ValueError("Expected int")
    return x + 10

safe_func = RunnableLambda(lambda x: int(x) + 5)

main = RunnableLambda(risky_func)
fallback_chain = main.with_fallbacks([safe_func])

print(fallback_chain.invoke("3"))  # 主失败 → 后备执行 → 输出 8

6. 重试机制（Retry）

自动重试失败操作，常用于网络请求或不稳定组件。

import time

counter = 0
def flaky_divide(x):
    global counter
    counter += 1
    print(f"Attempt {counter}")
    if counter < 3:
        raise Exception("Simulated failure")
    return x / 2

retriable = RunnableLambda(flaky_divide).with_retry(stop_after_attempt=3)
print(retriable.invoke(10))  # 第3次成功，输出 5.0

7. 条件分支（Conditional Chaining）

根据中间结果动态选择后续逻辑。

def route_by_value(x):
    if x > 10:
        return RunnableLambda(lambda _: "Large number")
    else:
        return x  # 直接返回值（非 Runnable）

chain = (
    RunnableLambda(lambda x: x + 5)  # 先加5
    | RunnableLambda(route_by_value)
)

print(chain.invoke(3))   # 3+5=8 ≤10 → 输出 8
print(chain.invoke(10))  # 10+5=15 >10 → 输出 "Large number"

⚠️ 注意：条件函数必须返回 Runnable 或可直接输出的值。

8. 生命周期监听（Listeners）

在组件执行前后插入回调，用于日志、监控、计时等。

import time
from langchain_core.tracers import Run

def on_start(run: Run):
    print(f"Started at: {run.start_time}")

def on_end(run: Run):
    print(f"Ended at: {run.end_time}, Duration: {run.end_time - run.start_time}")

slow_task = RunnableLambda(lambda x: time.sleep(x) or x * 2)
monitored_chain = slow_task.with_listeners(on_start=on_start, on_end=on_end)

monitored_chain.invoke(1)  # 打印开始/结束时间

基于上下文对话（持久化历史）

上下文记录必须存储在本地，并且要存放到磁盘当中，也就是数据库中。

from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import SQLChatMessageHistory

# 定义获取会话历史的函数
def get_session_history(session_id: str):
    return SQLChatMessageHistory(session_id, connection="sqlite:///chat.db")

# 构建带记忆的链
prompt_with_history = ChatPromptTemplate.from_messages([
    ("system", "你是一个幽默的程序员"),
    MessagesPlaceholder(variable_name="history"),
    ("user", "{input}")
])

base_chain = prompt_with_history | llm | StrOutputParser()

runnable = RunnableWithMessageHistory(
    base_chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history"
)

# 同一会话 ID 下的多次调用共享上下文
res1 = runnable.invoke(
    {"input": "中国一共有哪些直辖市？"},
    config={"configurable": {"session_id": "user123"}}
)
res2 = runnable.invoke(
    {"input": "这些城市中哪个最大？"},
    config={"configurable": {"session_id": "user123"}}
)