LangChain_从基础到高级

# chain.py - 最简单的链式操作
model = ChatZhipuAI(model="GLM-4-FlashX-250414")

prompt_template = ChatPromptTemplate.from_messages([
    ("system", "you are a facts expert who knows facts about {animal}."),
    ("human", "Tell me {fact_count} facts in chinese."),
])

# 🔗 核心：三步链式操作
chain = prompt_template | model | StrOutputParser() 

result = chain.invoke({"animal": "cat", "fact_count": 2})

输入数据 {"animal": "cat", "fact_count": 2}
    ↓
📝 prompt_template: 格式化提示模板
    ↓
🤖 model: AI模型生成回复
    ↓
📤 StrOutputParser(): 提取纯文本
    ↓
输出结果: "关于猫的两个事实..."

# chain2.py - 详细的链式操作分解
format_prompt = RunnableLambda(lambda x: prompt_template.format_prompt(**x))
invoke_model = RunnableLambda(lambda x: model.invoke(x.to_messages()))
parse_output = RunnableLambda(lambda x: x.content)

# 手动构建的链
chain = RunnableSequence(format_prompt, invoke_model, parse_output)

def debug_format_prompt(x):
    print(f"输入 x: {x}")                    # dict
    print(f"x 的类型: {type(x)}")            # <class 'dict'>
    result = prompt_template.format_prompt(**x)
    print(f"输出类型: {type(result)}")        # ChatPromptValue
    return result

# chain3.py - 多步骤处理链
# 第一个模板：生成动物事实
prompt_template = ChatPromptTemplate.from_messages([
    ("system", "you are a facts expert who knows facts about {animal}."),
    ("human", "Tell me {fact_count} facts in chinese."),
])

# 第二个模板：翻译内容
translation_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a translator and convert the following text into {language}."),
    ("human", "Translate the following text into {language}: {text}"),
])

# 数据转换函数
prepare_for_translation = RunnableLambda(
    lambda output: {"text": output, "language": "french"}
)

# 🔗 完整的多步骤链
chain = (
    prompt_template | 
    model | 
    StrOutputParser() | 
    prepare_for_translation | 
    translation_prompt | 
    model | 
    StrOutputParser()
)

{"animal": "cat", "fact_count": 2}
    ↓
📝 生成中文动物事实
    ↓
🤖 AI生成: "猫的两个事实..."
    ↓
📤 提取纯文本
    ↓
🔄 数据转换: {"text": "猫的两个事实...", "language": "french"}
    ↓
📝 翻译提示模板
    ↓
🤖 AI翻译: "Deux faits sur les chats..."
    ↓
📤 提取最终结果

# chain4.py - 并行处理示例
# 电影摘要模板
summary_template = ChatPromptTemplate.from_messages([
    ("system", "You are a movie critic."),
    ("human", "Provide a brief summary of the movie {movie_name}.")
])

# 分析函数
def analyze_plot(plot):
    plot_template = ChatPromptTemplate.from_messages([
        ("system", "You are a movie critic."),
        ("human", "Analyze the plot: {plot}. What are its strengths and weaknesses?"),
    ])
    return plot_template.format_prompt(plot=plot)

# 🔀 并行处理链
chain = (
    summary_template |
    model |
    StrOutputParser() |
    RunnableParallel({
        "summary": RunnableLambda(lambda x: x),  # 保留原始摘要
        "branches": RunnableParallel({
            "plot": plot_branch_chain,
            "characters": character_branch_chain
        })
    }) |
    RunnableLambda(combine_verdicts)
)

电影摘要文本
    ↓
┌─ RunnableParallel ────────────────────┐
│                                       │
│  📋 summary: 保留原始摘要              │
│                                       │
│  🔀 branches: RunnableParallel        │
│      ├─ 📈 plot: 剧情分析             │
│      └─ 👥 characters: 角色分析       │
│                                       │
└───────────────────────────────────────┘
    ↓
📊 combine_verdicts: 合并所有结果

# chain5.py - 条件分支处理
# 为不同情感定义不同的处理模板
positive_feedback_template = ChatPromptTemplate.from_messages([
    ("system", "你是一个乐于助人的助手。"),
    ("human", "为这条积极反馈生成一封感谢信：{feedback}。"),
])

negative_feedback_template = ChatPromptTemplate.from_messages([
    ("system", "你是一个乐于助人的助手。"),
    ("human", "为这条负面反馈生成一个回复：{feedback}。"),
])

# 🌲 条件分支路由
branches = RunnableBranch(
    (
        lambda x: "positive" in x,
        positive_feedback_template | model | StrOutputParser()
    ),
    (
        lambda x: "negative" in x,
        negative_feedback_template | model | StrOutputParser()
    ),
    (
        lambda x: "neutral" in x,
        neutral_feedback_template | model | StrOutputParser()
    ),
    escalate_feedback_template | model | StrOutputParser()  # 默认分支
)

# 🔗 完整的智能路由链
chain = classification_chain | branches

用户反馈
    ↓
🔍 情感分类: "这个产品很糟糕..."
    ↓
📊 分类结果: "negative"
    ↓
🌲 RunnableBranch 决策:
    ├─ ✅ positive? → 感谢信模板
    ├─ ✅ negative? → 问题解决模板 ← 选中
    ├─ ❌ neutral? → 详情询问模板
    └─ ❌ default → 人工客服模板
    ↓
📝 生成针对性回复

# ✅ 推荐：清晰的链式结构
chain = prompt | model | parser

# ✅ 推荐：适当的错误处理
try:
    result = chain.invoke(input_data)
except Exception as e:
    print(f"处理失败: {e}")

# ✅ 推荐：调试功能
debug_chain = chain.with_config({"run_name": "debug_run"})