第一篇：基础认知#

python --version  # 需要 Python 3.10 或更高版本

# 使用 pip 安装最新版本
pip install langchain langchain-openai langgraph langchain-community

# 或使用 uv (推荐)
uv pip install langchain langchain-openai langgraph langchain-community

# 如需指定版本（推荐使用1.0.7或更高版本）
pip install langchain>=1.0.7 langchain-openai>=1.0.3 langgraph>=1.0.3

# 创建 .env 文件
OPENAI_API_KEY=sk-your-api-key-here
LANGSMITH_API_KEY=your-langsmith-key  # 可选,用于监控
LANGSMITH_TRACING=true  # 可选

# 在代码中加载
from dotenv import load_dotenv
import os

load_dotenv()

# 验证环境变量
required_vars = ["OPENAI_API_KEY"]
for var in required_vars:
    if not os.getenv(var):
        raise EnvironmentError(f"缺少必需的环境变量: {var}")

# pyproject.toml 推荐配置
[tool.poetry.dependencies]
python = "^3.10"
langchain = "^1.0.7"
langchain-openai = "^1.0.3"
langgraph = "^1.0.3"
langchain-community = "^0.3.0"
langchain-core = "^1.0.7"
langsmith = "^0.4.43"
python-dotenv = "^1.0.0"

# requirements.txt 格式
# langchain>=1.0.7
# langchain-openai>=1.0.3
# langgraph>=1.0.3
# langchain-community>=0.3.0
# langchain-core>=1.0.7
# langsmith>=0.4.43
# python-dotenv>=1.0.0

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

# 步骤1: 定义工具
@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气"""
    return f"{city}今天天气晴朗，温度25°C"

@tool
def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        result = eval(expression)
        return f"计算结果: {result}"
    except Exception as e:
        return f"计算错误: {str(e)}"

# 步骤2: 创建 Agent
agent = create_agent(
    model=ChatOpenAI(model="gpt-4"),
    tools=[get_weather, calculate],
    system_prompt="你是一个有帮助的助手，可以查询天气和进行计算。"
)

# 步骤3: 运行 Agent
result = agent.invoke({
    "messages": [("user", "北京天气如何？另外帮我算一下 25 * 4")]
})

# 查看结果
print(result["messages"][-1].content)

北京今天天气晴朗，温度25°C。
25 * 4 的计算结果是 100。

from langchain_core.runnables import Runnable

# 所有组件均实现 Runnable 接口
class Runnable:
    def invoke(self, input, config=None): ...       # 同步调用
    def ainvoke(self, input, config=None): ...      # 异步调用
    def stream(self, input, config=None): ...       # 流式输出
    def astream(self, input, config=None): ...      # 异步流式
    def batch(self, inputs, config=None): ...       # 批量处理

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

# 所有组件都是 Runnable
prompt = ChatPromptTemplate.from_template("Tell me a joke about {topic}")
model = ChatOpenAI()
parser = StrOutputParser()

# 统一的调用方式
result = prompt.invoke({"topic": "AI"})
result = model.invoke("Tell me a joke")
result = parser.invoke("some text")

from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4")

# 同步调用
response = model.invoke("What is LangChain?")
print(response.content)

# 流式输出
for chunk in model.stream("Tell me a long story"):
    print(chunk.content, end="", flush=True)

# 批量处理（自动并发优化）
inputs = [
    "What is AI?",
    "What is ML?",
    "What is LLM?"
]

results = model.batch(inputs)
for result in results:
    print(result.content)

import asyncio
from langchain_openai import ChatOpenAI

model = ChatOpenAI()

async def async_batch_example():
    inputs = [
        "What is AI?",
        "What is ML?",
        "What is LLM?",
        "What is NLP?",
        "What is DL?"
    ]

    # 异步批量处理
    results = await model.abatch(inputs)

    for i, result in enumerate(results):
        print(f"Result {i+1}: {result.content}")

# 运行异步任务
asyncio.run(async_batch_example())

import asyncio
from langchain_openai import ChatOpenAI

model = ChatOpenAI()

async def main():
    # 异步调用
    response = await model.ainvoke("What is async programming?")
    print(response.content)

asyncio.run(main())

async def stream_example():
    async for chunk in model.astream("Tell me a story"):
        print(chunk.content, end="", flush=True)

asyncio.run(stream_example())

# ❌ 同步方式（串行执行，慢）
def sync_batch():
    results = []
    for query in queries:
        results.append(model.invoke(query))
    return results

# ✅ 异步方式（并发执行，快）
async def async_batch():
    tasks = [model.ainvoke(query) for query in queries]
    return await asyncio.gather(*tasks)

from langchain_core.runnables import RunnableLambda

def uppercase(text: str) -> str:
    return text.upper()

# 包装为 Runnable
runnable_upper = RunnableLambda(uppercase)

# 统一调用方式
result = runnable_upper.invoke("hello")  # "HELLO"

from langchain_core.runnables import RunnableParallel

parallel = RunnableParallel(
    joke=ChatPromptTemplate.from_template("Tell a joke about {topic}") | model,
    poem=ChatPromptTemplate.from_template("Write a poem about {topic}") | model
)

# 并行执行
result = parallel.invoke({"topic": "AI"})
print(result["joke"])
print(result["poem"])

from langchain_core.runnables import RunnableBranch

branch = RunnableBranch(
    (lambda x: len(x) > 100, long_text_handler),
    (lambda x: len(x) > 10, medium_text_handler),
    short_text_handler  # 默认分支
)

result = branch.invoke("some text")

from langchain_openai import ChatOpenAI

primary_model = ChatOpenAI(model="gpt-4")
fallback_model = ChatOpenAI(model="gpt-3.5-turbo")

# 直接使用 with_fallbacks 方法，无需导入额外类
model_with_fallback = primary_model.with_fallbacks([fallback_model])

# 如果 GPT-4 失败，自动使用 GPT-3.5
result = model_with_fallback.invoke("Hello")

# ✅ 示例1: 只对特定异常类型执行降级
model_with_fallback = primary_model.with_fallbacks(
    fallbacks=[fallback_model],  # ✅ 官方标准参数：fallbacks (复数，列表)
    exceptions_to_handle=(TimeoutError, ConnectionError),  # ✅ 官方标准参数
)

# ✅ 示例2: 将异常信息传递给备用方案
from langchain_core.runnables import RunnableLambda

def handle_with_error_context(inputs):
    """备用方案可以访问异常信息"""
    if "error" in inputs:
        print(f"Original error: {inputs['error']}")
    return fallback_model.invoke(inputs["input"])

model_with_error_context = primary_model.with_fallbacks(
    fallbacks=[RunnableLambda(handle_with_error_context)],  # ✅ 使用 fallbacks 参数名
    exception_key="error"  # ✅ 官方标准参数：异常会以 "error" 键传递
)

# ⚠️ 重要：使用 exception_key 时，主 Runnable 和所有备用方案都必须接受字典输入
result = model_with_error_context.invoke({"input": "Hello"})

def with_fallbacks(
    self,
    fallbacks: Sequence[Runnable[Input, Output]],  # 必需
    *,
    exceptions_to_handle: Tuple[Type[BaseException], ...] = (Exception,),  # 可选
    exception_key: Optional[str] = None  # 可选
) -> RunnableWithFallbacksT[Input, Output]:
    ...

# ❌ 命令式（手动控制流程）
def imperative_chain(input):
    step1_result = prompt.invoke(input)
    step2_result = model.invoke(step1_result)
    step3_result = parser.invoke(step2_result)
    return step3_result

# ✅ 声明式（LCEL 管道）
chain = prompt | model | parser
result = chain.invoke(input)

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

# 管道组合
chain = (
    ChatPromptTemplate.from_template("Tell me about {topic}")
    | ChatOpenAI()
    | StrOutputParser()
)

# 自动按顺序执行
result = chain.invoke({"topic": "LangChain"})

from langchain_core.runnables import RunnablePassthrough

chain = {
    "context": retriever | format_docs,
    "question": RunnablePassthrough()
} | prompt | model

# context 和 question 并行处理
result = chain.invoke("What is LangChain?")

from langchain_core.runnables import RunnablePassthrough
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 创建向量检索器
vectorstore = Chroma.from_texts(
    ["LangChain是一个AI应用框架", "它支持RAG和Agent"],
    embedding=OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever()

# 使用 assign() 添加检索上下文
chain = (
    RunnablePassthrough.assign(
        context=retriever  # 添加 context 字段
    )
    | ChatPromptTemplate.from_template(
        "基于以下上下文回答问题:\n{context}\n\n问题: {question}"
    )
    | ChatOpenAI()
    | StrOutputParser()
)

# 输入只需要 question，context 会自动添加
result = chain.invoke({"question": "什么是LangChain?"})
# 内部流程: {"question": "..."} -> {"question": "...", "context": [...]}

# 多次 assign 叠加字段
chain = (
    RunnablePassthrough.assign(
        context=retriever  # 添加检索结果
    )
    .assign(
        context_count=lambda x: len(x["context"])  # 添加统计信息
    )
    .assign(
        timestamp=lambda x: "2025-11-17"  # 添加时间戳
    )
    | prompt
    | model
)

# 输入: {"question": "..."}
# 第一步后: {"question": "...", "context": [...]}
# 第二步后: {"question": "...", "context": [...], "context_count": 3}
# 第三步后: {"question": "...", "context": [...], "context_count": 3, "timestamp": "..."}

# 场景1: RAG 添加检索上下文
rag_chain = (
    RunnablePassthrough.assign(context=retriever)
    | rag_prompt
    | model
)

# 场景2: 添加多个数据源
multi_source_chain = (
    RunnablePassthrough.assign(
        docs=doc_retriever,
        history=history_retriever,
        metadata=metadata_fetcher
    )
    | prompt
    | model
)

# 场景3: 数据转换
transform_chain = (
    RunnablePassthrough.assign(
        upper_text=lambda x: x["text"].upper(),
        word_count=lambda x: len(x["text"].split())
    )
    | processor
)

# 简单顺序
chain = step1 | step2 | step3

# 复杂顺序
chain = (
    {"input": RunnablePassthrough()}
    | prompt
    | model
    | {"output": parser, "raw": RunnablePassthrough()}
)

# 并行获取多个信息
chain = RunnableParallel(
    summary=summarize_chain,
    keywords=extract_keywords_chain,
    sentiment=sentiment_chain
)

from langchain_core.runnables import RunnableBranch

# 根据输入长度选择不同处理
chain = RunnableBranch(
    (lambda x: len(x["text"]) > 1000, long_text_chain),
    (lambda x: len(x["text"]) > 100, medium_text_chain),
    short_text_chain
)

# 使用 RunnableLambda 实现循环
def iterative_refine(input):
    result = input
    for _ in range(3):
        result = refine_chain.invoke(result)
    return result

chain = RunnableLambda(iterative_refine)

# 多级降级
chain = (
    primary_model
    .with_fallbacks(fallbacks=[backup_model_1, backup_model_2])  # ✅ 使用 fallbacks 参数名
)

# 只对特定异常执行降级
chain = (
    primary_model
    .with_fallbacks(
        fallbacks=[backup_model_1, backup_model_2],  # ✅ 使用 fallbacks 参数名
        exceptions_to_handle=(TimeoutError, ConnectionError)  # ✅ 官方标准参数
    )
)

# 直接使用 with_retry 方法，无需单独导入
chain = (
    prompt | model | parser
).with_retry(
    stop_after_attempt=3,  # 最大重试次数
    wait_exponential_jitter=True,  # 指数退避 + 随机抖动
    retry_if_exception_type=(Exception,)  # 指定需要重试的异常类型
)

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

# 只对特定异常重试
chain = (prompt | model | parser).with_retry(
    retry_if_exception_type=(TimeoutError, ConnectionError),
    stop_after_attempt=5,
    wait_exponential_jitter=True
)

# 禁用指数退避（立即重试）
chain = (prompt | model | parser).with_retry(
    stop_after_attempt=3,
    wait_exponential_jitter=False  # 禁用指数退避，立即重试
)

from langchain_openai import ChatOpenAI

# ✅ 正确：在模型构造时设置timeout
model = ChatOpenAI(
    model="gpt-4",
    timeout=30,  # 30秒超时
    max_retries=2
)

chain = prompt | model | parser
result = chain.invoke(input)

from langchain_openai import ChatOpenAI

# 主模型：严格超时
slow_model = ChatOpenAI(model="gpt-4", timeout=10)

# 降级模型：快速响应
fast_model = ChatOpenAI(model="gpt-3.5-turbo", timeout=5)

# 组合超时 + 降级
chain = (prompt | slow_model | parser).with_fallbacks([
    prompt | fast_model | parser
])

from langchain_core.runnables import RunnableConfig

# RunnableConfig支持的参数
result = chain.invoke(
    input,
    config=RunnableConfig(
        max_concurrency=5,      # 最大并发数
        tags=["production"],     # 标签（用于监控）
        metadata={"user": "alice"}  # 元数据
    )
)

from langchain_core.caches import InMemoryCache
from langchain_core.globals import set_llm_cache

# 启用缓存
set_llm_cache(InMemoryCache())

# 相同请求直接返回缓存结果
model.invoke("What is AI?")  # 调用 LLM
model.invoke("What is AI?")  # 返回缓存（不调用 LLM）

# 批处理优化（使用max_concurrency控制并发）
chain = prompt | model.with_config({"max_concurrency": 10})

# 内部自动合并请求
results = chain.batch(inputs)

# 流式传输减少延迟
for chunk in chain.stream(input):
    print(chunk, end="")