第六章：LangChain/AutoGen框架实战

---

6.1 LangChain核心组件与实战

6.1.1 LangChain架构概览与核心概念

先说一个真实的痛点。

2022年底，LangChain刚出来那会儿，如果你要让大模型"接入"自己的私有数据，基本上是手写一切：

• 自己写向量相似度计算
• 自己拼Prompt模板
• 自己管理对话历史
• 自己调用OpenAI API、处理重试、处理限流
• 自己解析Function Calling的输出
• 自己把工具执行结果塞回Prompt

一套下来，几百行代码，换个模型就得大面积改。更别提要接RAG、要接Agent、要换向量数据库……

LangChain要解决的就是这个问题——把大模型应用开发中那些"反复要写的脏活累活"抽象成标准组件，像搭积木一样拼出应用。

---

LangChain的核心设计哲学

LangChain的命名其实说清楚了它的定位：把大模型（Chain）和各种外部能力"链"起来。

它的架构围绕6个核心模块展开，这6个模块几乎覆盖了所有LLM应用的需求：

┌────────────────────────────────────────────────┐
│                 LangChain 架构                │
│                                            │
│  ┌──────────┐  ┌──────────┐  ┌────────┐│
│  │ Model I/O │  │  Prompts  │  │ Chains ││
│  │  模型交互  │  │  提示词模板 │  │ 工作流  ││
│  └──────────┘  └──────────┘  └────────┘│
│         ↓            ↓            ↓          │
│  ┌──────────┐  ┌──────────┐  ┌────────┐│
│  │ Retrieval  │  │   Agents  │  │ Memory ││
│  │  检索增强  │  │  智能体    │  │ 记忆  ││
│  └──────────┘  └──────────┘  └────────┘│
└────────────────────────────────────────────────┘

用一句话理解：LangChain = 大模型应用的"标准库"，就像Python标准库让你可以不依赖第三方包就写出功能完整的程序，LangChain让你不依赖大量手写代码就能搭出LLM应用。

---

LangChain的发展演变

这里有必要提一下版本演变，因为网上很多教程是基于旧版API写的，跑不动。

• LangChain v0.0.x（2022-2023）：最早版本，API设计比较随意
• LangChain v0.1.x（2023）：引入了LCEL（LangChain Expression Language），用管道操作符 | 把组件串起来
• LangChain v0.2.x（2024至今）：进一步统一了API，强化了Runnable接口

本书所有代码示例基于 v0.2.x+，如果你在跑代码时遇到API报错，先检查版本。

# 安装指定版本
pip install langchain>=0.2.0 langchain-openai langchain-community

---

核心概念：Runnable接口

从v0.1开始，LangChain所有组件都实现了 Runnable 接口，这是理解现代LangChain的关键。

Runnable 的核心方法：

# 每个Runnable都支持这四种调用方式
runnable.invoke(input)        # 同步调用，返回最终结果
runnable.ainvoke(input)      # 异步调用
runnable.batch([inp1, inp2]) # 批量调用
runnable.stream(input)        # 流式输出

更重要的是，Runnable支持管道操作符 |，让你可以把多个组件像Unix管道一样串起来：

# 这就是LCEL（LangChain Expression Language）
chain = prompt | llm | output_parser
result = chain.invoke({"question": "什么是RAG?"})

这个设计非常优雅——prompt的输出自动变成llm的输入，llm的输出自动变成output_parser的输入，不需要手写中间变量。

---

为什么需要这个技术？ 总结一句话：

LangChain把LLM应用开发从"手写每个细节"提升到了"组合标准组件"的层次，大幅降低了重复代码量，也提升了可维护性。

---

6.1.2 Model I/O：模型交互标准化

痛点再描述一下。

如果你直接调OpenAI的API，代码大概是这样：

import openai
response = openai.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "你好"}],
    temperature=0.7,
    max_tokens=1000
)
print(response.choices[0].message.content)

这看起来没问题，但一旦你要：

• 换模型（从GPT-4换到Claude，或者换到开源的Qwen）
• 批量处理100条数据
• 接入RAG（需要把检索结果拼进Prompt）
• 做流式输出

原来的代码就要大面积改动。

LangChain的Model I/O层就是来解决这个"模型接入碎片化"的问题的。

---

Model I/O三层架构

┌──────────────────────────────────────┐
│            Model I/O               │
│                                  │
│  ┌────────────────────────────┐   │
│  │       Chat Models          │   │
│  │  （对话模型，最常用）      │   │
│  └────────────────────────────┘   │
│            ↓                       │
│  ┌────────────────────────────┐   │
│  │       LLM（Text Models）  │   │
│  │  （纯文本补全模型）        │   │
│  └────────────────────────────┘   │
│            ↓                       │
│  ┌────────────────────────────┐   │
│  │     Embeddings            │   │
│  │  （文本向量化模型）       │   │
│  └────────────────────────────┘   │
└──────────────────────────────────────┘

---

Chat Models：对话模型（最常用）

现代大模型应用几乎都是基于对话格式（Messages）的，LangChain用 ChatModel 统一了所有模型的调用接口。

代码实战6.1：统一调用多种大模型

"""
代码实战6.1：使用LangChain的ChatModel统一调用多种大模型
无论底层是OpenAI、Claude还是开源模型，代码写法完全一致
"""

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_community.chat_models import ChatZhipuAI  # 智谱AI
from langchain_core.messages import HumanMessage, SystemMessage
from typing import List, Dict, Any


def demo_chat_model_unified():
    """
    演示如何用统一的接口调用不同的大模型
    核心思想：换了模型，业务代码不需要改
    """

    # ========== 方式1：OpenAI GPT-4 ==========
    # 只需要改这一行，后面的代码完全不用动
    chat_model = ChatOpenAI(
        model="gpt-4-turbo",
        temperature=0.7,
        max_tokens=1000,
        # api_key="sk-..."  # 也可以通过环境变量OPENAI_API_KEY设置
    )

    # 构造消息列表（标准格式，所有模型都认这个格式）
    messages = [
        SystemMessage(content="你是一个有帮助的AI助手，回答简洁明了。"),
        HumanMessage(content="请用一句话解释什么是RAG")
    ]

    # 调用模型（同步）
    response = chat_model.invoke(messages)
    print("GPT-4 回复：")
    print(response.content)
    print("-" * 50)

    # ========== 方式2：切换到Anthropic Claude（只需要改模型初始化）==========

    # 把上面那一行的ChatOpenAI换成ChatAnthropic
    # chat_model = ChatAnthropic(
    #     model="claude-3-opus-20240229",
    #     temperature=0.7,
    #     max_tokens=1000
    # )
    # 后面的 messages = [...] 和 response = chat_model.invoke(messages) 完全不用改！

    return response.content


def demo_streaming_output():
    """
    演示流式输出：大模型边生成边返回，用户体验更好
    所有ChatModel都支持stream方法
    """
    chat_model = ChatOpenAI(model="gpt-3.5-turbo", streaming=True)

    messages = [HumanMessage(content="从1数到10，每个数字一行")]

    print("流式输出演示：")
    for chunk in chat_model.stream(messages):
        # chunk.content 是每个流式片段的内容
        if chunk.content:
            print(chunk.content, end="", flush=True)
    print()  # 换行


def demo_async_batch():
    """
    演示批量异步调用：同时处理多个请求，大幅提升吞吐量
    适合批量数据处理场景
    """
    import asyncio

    chat_model = ChatOpenAI(model="gpt-3.5-turbo")

    messages_list = [
        [HumanMessage(content="解释一下深度学习")],
        [HumanMessage(content="解释一下强化学习")],
        [HumanMessage(content="解释一下迁移学习")]
    ]

    # 批量调用（自动并发）
    responses = chat_model.batch(messages_list)

    for i, response in enumerate(responses):
        print(f"任务{i+1}结果（前50字）：{response.content[:50]}...")


if __name__ == "__main__":
    print("=" * 60)
    print("代码实战6.1：统一调用多种大模型")
    print("=" * 60)
    demo_chat_model_unified()
    # demo_streaming_output()  # 取消注释可测试流式输出

关键点：invoke() 方法返回的是一个 AIMessage 对象，不只是字符串。它包含：

response.content          # 模型生成的文字内容（最常用的字段）
response.additional_kwargs  # 额外信息（如token使用量）
response.response_metadata # 响应元数据

---

LLM（Text Models）：纯文本补全模型

这是更早期的接口，用于那些只支持"纯文本补全"的模型（如GPT-2、部分开源模型）。现在大多数场景直接用Chat Models就行。

from langchain_openai import OpenAI  # 注意：这是文本补全接口，不是Chat！

llm = OpenAI(model="gpt-3.5-turbo-instruct")
result = llm.invoke("解释什么是深度学习")
print(result)

---

Embeddings：文本向量化模型

RAG应用中，把文本转成向量是核心步骤。LangChain用 Embeddings 接口统一了所有嵌入模型。

代码实战6.2：使用Embeddings接口统一调用嵌入模型

"""
代码实战6.2：使用LangChain的Embeddings接口
支持OpenAI、HuggingFace、本地模型等多种嵌入模型
"""

from langchain_openai import OpenAIEmbeddings
from langchain_community.embeddings import HuggingFaceEmbeddings
from typing import List
import numpy as np


def demo_embeddings_comparison():
    """
    对比不同嵌入模型的使用方式
    核心发现：接口完全一致，换模型只需要改初始化那一行
    """

    # ========== 方式1：OpenAI嵌入模型 ==========
    # 需要网络，需要API Key
    openai_embeddings = OpenAIEmbeddings(
        model="text-embedding-3-small",  # 推荐用这个，性价比高
        # dimensions=1024  # 可以指定输出维度（节省存储和计算）
    )

    texts = ["深度学习是机器学习的一个分支", "大语言模型改变了AI的格局"]

    # embed_documents: 批量处理多个文本（推荐，省API调用次数）
    embeddings = openai_embeddings.embed_documents(texts)
    print(f"OpenAI嵌入维度：{len(embeddings[0])}")
    print(f"生成了 {len(embeddings)} 个向量")

    # embed_query: 专门处理查询文本（有些模型对查询和文档用不同的编码方式）
    query_embedding = openai_embeddings.embed_query("什么是深度学习？")
    print(f"查询向量维度：{len(query_embedding)}")

    # ========== 方式2：HuggingFace开源嵌入模型（本地运行，无需API Key）==========
    # 适合对数据隐私要求高的场景
    # 首次运行会自动下载模型（约400MB）
    hf_embeddings = HuggingFaceEmbeddings(
        model_name="BAAI/bge-large-zh-v1.5",  # 中文效果很好的开源模型
        model_kwargs={"device": "cpu"},  # 有GPU的话改成 "cuda"
        encode_kwargs={"normalize_embeddings": True}  # 归一化，使点积=余弦相似度
    )

    hf_embeddings_result = hf_embeddings.embed_documents(texts)
    print(f"\nBGE嵌入维度：{len(hf_embeddings_result[0])}")

    # 计算两个文本的相似度（余弦相似度）
    vec1 = np.array(hf_embeddings_result[0])
    vec2 = np.array(hf_embeddings_result[1])
    cosine_sim = np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))
    print(f"两个文本的余弦相似度：{cosine_sim:.4f}")

    return embeddings


if __name__ == "__main__":
    print("=" * 60)
    print("代码实战6.2：统一调用嵌入模型")
    print("=" * 60)
    demo_embeddings_comparison()

---

模型调用过程中的错误处理与重试

生产环境中，模型API调用可能因为网络抖动、限流、超时等原因失败。LangChain内置了重试机制：

from langchain_openai import ChatOpenAI

# 配置自动重试
chat_model = ChatOpenAI(
    model="gpt-4-turbo",
    max_retries=3,          # 最多重试3次
    request_timeout=30,       # 单次请求超时30秒
    # LangChain会自动处理：网络错误、速率限制错误等
)

更精细的控制可以用 tenacity 库自己包装，但大多数场景用上面内置的参数就够了。

---

6.1.3 Retrieval：构建RAG应用

Retrieval是LangChain中专门用于RAG（检索增强生成） 的模块。它把"检索"抽象成了标准组件，可以和ChatModel、Chain无缝集成。

---

RAG在LangChain中的完整数据流转

用户提问
   ↓
【RetrievalStep 1】加载文档（Document Loader）
   ↓
【RetrievalStep 2】文档分块（Text Splitter）
   ↓
【RetrievalStep 3】文本向量化（Embeddings）
   ↓
【RetrievalStep 4】存入向量数据库（VectorStore）
   ↓
【查询时】用户问题向量化 → 相似度搜索 → 取Top-K个相关文档
   ↓
【RetrievalStep 5】将相关文档拼进Prompt
   ↓
【RetrievalStep 6】发给LLM生成答案
   ↓
返回最终答案

---

Step 1 & 2：加载文档与分块

LangChain提供了大量的DocumentLoader，支持PDF、Word、Markdown、网页、Notion等格式。

代码实战6.3：从零构建RAG检索系统

"""
代码实战6.3：使用LangChain构建完整的RAG检索系统
包含：文档加载 → 分块 → 向量化 → 存储 → 检索
"""

from langchain_community.document_loaders import TextLoader, PDFMinerLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from typing import List, Dict
import os


def build_rag_retrieval_pipeline(
    file_path: str,
    persist_directory: str = "./chroma_db"
):
    """
    构建RAG检索流水线的完整流程

    Args:
        file_path: 文档路径（支持.txt, .pdf等）
        persist_directory: 向量数据库持久化目录

    Returns:
        retriever: 检索器对象，可以直接用来检索相关文档
    """

    # ========== Step 1：加载文档 ==========
    print(f"Step 1：加载文档 {file_path} ...")

    if file_path.endswith(".txt"):
        loader = TextLoader(file_path, encoding="utf-8")
    elif file_path.endswith(".pdf"):
        loader = PDFMinerLoader(file_path)
    else:
        raise ValueError(f"不支持的文件格式：{file_path}")

    documents = loader.load()
    print(f"  加载完成，共 {len(documents)} 个文档")

    # ========== Step 2：文档分块 ==========
    print("Step 2：文档分块...")

    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,           # 每个块约500字符
        chunk_overlap=100,         # 块之间重叠100字符（避免语义断裂）
        length_function=len,
        separators=["\n\n", "\n", "。", "；", "，", " ", ""]  # 优先在句子边界切分
    )

    splits = text_splitter.split_documents(documents)
    print(f"  分块完成，共 {len(splits)} 个块")
    print(f"  块大小统计：平均约 {sum(len(s.page_content) for s in splits) // len(splits)} 字符")

    # ========== Step 3：文本向量化 ==========
    print("Step 3：初始化嵌入模型...")

    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

    # ========== Step 4：存入向量数据库 ==========
    print(f"Step 4：存入向量数据库（持久化到 {persist_directory}）...")

    # Chroma会自动处理：向量化 + 存储 + 建立索引
    vectorstore = Chroma.from_documents(
        documents=splits,
        embedding=embeddings,
        persist_directory=persist_directory
    )

    print(f"  存储完成，数据库中共有 {vectorstore._collection.count()} 个向量")

    # ========== 返回检索器（Retriever）==========
    # Retriever是LangChain中对"检索"的标准抽象
    retriever = vectorstore.as_retriever(
        search_type="similarity",  # 相似度搜索
        search_kwargs={"k": 3}     # 返回最相关的3个块
    )

    print("  RAG检索流水线构建完成！")
    return retriever


def demo_retrieval(retriever, query: str):
    """
    演示如何使用检索器
    """
    print(f"\n检索查询：{query}")
    docs = retriever.invoke(query)
    print(f"检索到 {len(docs)} 个相关文档块：")
    for i, doc in enumerate(docs, 1):
        print(f"\n  块{i}（前100字）：{doc.page_content[:100]}...")
        print(f"  元数据：{doc.metadata}")


if __name__ == "__main__":
    print("=" * 60)
    print("代码实战6.3：构建RAG检索系统")
    print("=" * 60)

    # 注意：需要准备一个测试文档，或者用手动构造的文档来测试
    # 这里演示用字符串构造文档
    from langchain_core.documents import Document

    test_docs = [
        Document(page_content="RAG（Retrieval-Augmented Generation）是一种结合检索和生成的技术。它先从外部知识库检索相关信息，然后基于检索到的内容生成答案。", metadata={"source": "RAG介绍"}),
        Document(page_content="LangChain是一个用于开发LLM应用的框架。它提供了ChatModel、Chain、Agent、Retrieval等核心组件。", metadata={"source": "LangChain介绍"}),
        Document(page_content="向量数据库用于存储文本的向量表示，支持高效的相似度搜索。常见的向量数据库有Chroma、Pinecone、Milvus等。", metadata={"source": "向量数据库介绍"}),
    ]

    # 跳过文件加载，直接用测试文档演示检索
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vectorstore = Chroma.from_documents(
        documents=test_docs,
        embedding=embeddings,
        persist_directory="./test_chroma_db"
    )
    retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

    demo_retrieval(retriever, "什么是RAG？")

---

Step 5 & 6：将检索结果接入LLM（RAG Chain）

有了Retriever，接下来要把"检索"和"生成"连起来，形成完整的RAG链路。LangChain用 create_retrieval_chain 或者LCEL来优雅地实现这个。

代码实战6.4：构建端到端RAG Chain

"""
代码实战6.4：使用LCEL构建端到端RAG Chain
将检索器和ChatModel用管道操作符连接起来
"""

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from langchain_core.documents import Document


def format_docs(docs):
    """
    将检索到的文档块格式化为字符串
    这是RAG Prompt中context的来源
    """
    return "\n\n".join(doc.page_content for doc in docs)


def build_rag_chain():
    """
    使用LCEL构建RAG Chain
    数据流：问题 → 检索相关文档 → 填入Prompt → LLM生成答案
    """

    # ========== 准备检索器（复用代码实战6.3的逻辑）==========
    test_docs = [
        Document(page_content="RAG（Retrieval-Augmented Generation）是一种结合检索和生成的技术。它先从外部知识库检索相关信息，然后基于检索到的内容生成答案。RAG可以有效减少大模型的幻觉问题。", metadata={"source": "RAG介绍"}),
        Document(page_content="LangChain是一个用于开发LLM应用的框架。它提供了ChatModel、Chain、Agent、Retrieval等核心组件。LangChain支持多种大模型，包括OpenAI、Anthropic、HuggingFace等。", metadata={"source": "LangChain介绍"}),
        Document(page_content="向量数据库用于存储文本的向量表示，支持高效的相似度搜索。常见的向量数据库有Chroma（轻量级，适合本地开发）、Pinecone（云服务）、Milvus（高性能分布式）。", metadata={"source": "向量数据库介绍"}),
        Document(page_content="Agent是大模型应用的重要方向。Agent可以自主规划、调用工具、执行动作。主流的Agent范式包括ReAct、Plan-and-Execute、Tree-of-Thought等。", metadata={"source": "Agent介绍"}),
    ]

    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vectorstore = Chroma.from_documents(
        documents=test_docs,
        embedding=embeddings,
        persist_directory="./test_chroma_db"
    )
    retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

    # ========== 准备LLM ==========
    llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

    # ========== 构建Prompt模板 ==========
    # 这是RAG的核心Prompt，格式直接影响生成质量
    prompt = ChatPromptTemplate.from_template("""
请基于以下参考文档回答用户的问题。
如果参考文档中没有相关信息，请明确说明"根据现有资料无法回答"，不要编造答案。

【参考文档】
{context}

【用户问题】
{question}

【回答要求】
1. 答案要基于参考文档中的内容
2. 语言简洁明了
3. 如果信息不足，明确说明
""".strip())

    # ========== 使用LCEL构建RAG Chain ==========
    # 这是整段代码最核心的部分，理解数据是怎么流的：
    #
    # 输入：{"question": "用户问题"}
    #   ↓
    # RunnablePassthrough.assign: 在输入字典中新增一个"context"键
    #   - context的值 = retriever.invoke(question) → format_docs → 格式化字符串
    #   ↓
    # prompt: 把{"question": ..., "context": ...} 填入Prompt模板
    #   ↓
    # llm: 把Prompt发给LLM
    #   ↓
    # StrOutputParser: 把LLM的AIMessage转成纯字符串
    #
    # 用LCEL的管道操作符 | 把这些步骤串起来

    rag_chain = (
        {
            "context": retriever | format_docs,  # 检索相关文档并格式化
            "question": RunnablePassthrough()       # 原样传递用户问题
        }
        | prompt
        | llm
        | StrOutputParser()
    )

    return rag_chain, retriever


def demo_rag_chain(rag_chain):
    """演示RAG Chain的效果"""
    questions = [
        "什么是RAG？",
        "LangChain支持哪些大模型？",
        "有哪些常见的向量数据库？",
        "Agent有哪些主流范式？"  # 这个问题在参考文档里有答案
    ]

    for question in questions:
        print(f"\n问题：{question}")
        answer = rag_chain.invoke({"question": question})
        print(f"答案：{answer}")


if __name__ == "__main__":
    print("=" * 60)
    print("代码实战6.4：端到端RAG Chain")
    print("=" * 60)

    rag_chain, retriever = build_rag_chain()
    demo_rag_chain(rag_chain)

这段代码是整个RAG部分最核心的实战代码。理解LCEL的数据流是理解现代LangChain的关键：

输入字典 {"question": "什么是RAG？"}
    ↓
retriever | format_docs
    → 执行检索，拿到相关文档，格式化成字符串
    → 结果存入字典的 "context" 键
    ↓
{"question": "...", "context": "RAG是..."}
    ↓
prompt.format(question=..., context=...)
    → 生成最终Prompt字符串
    ↓
llm.invoke(prompt)
    → 生成AIMessage
    ↓
StrOutputParser().parse(ai_message)
    → 提取出纯文本字符串
    ↓
返回最终答案

---

6.1.4 Chain：构建可组合的工作流

Chain是LangChain最早的核心概念，字面意思就是"把多个处理步骤链起来"。

在LCEL出现之前，Chain是用特定的类（如LLMChain、ConversationChain）来定义的。现在更推荐使用LCEL来构建Chain，因为它更灵活、更统一。

---

几种常见的Chain模式

模式1：最简单的Prompt → LLM Chain

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

# 用LCEL构建最简单的Chain
prompt = ChatPromptTemplate.from_template("{topic}领域的三个关键技术是什么？")
llm = ChatOpenAI(model="gpt-3.5-turbo")
output_parser = StrOutputParser()

chain = prompt | llm | output_parser
result = chain.invoke({"topic": "大模型"})

模式2：多步Chain（上一步的输出是下一步的输入）

from langchain_core.runnables import RunnablePassthrough

# 场景：先生成大纲，再基于大纲生成完整文章
outline_prompt = ChatPromptTemplate.from_template("请为"{topic}"这个主题生成一个文章大纲，包含3个章节")
article_prompt = ChatPromptTemplate.from_template("基于以下大纲，写一篇详细文章：\n\n{outline}")

llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)

# Chain 1：生成大纲
outline_chain = outline_prompt | llm | StrOutputParser()

# Chain 2：基于大纲生成文章（把Chain 1的输出接进来）
# RunnablePassthrough() 把输入原样传递，这里用来把outline_chain的输出接给article_prompt
article_chain = (
    outline_chain                          # 第一步：生成大纲
    | (lambda outline: {"outline": outline})  # 把大纲包装成字典
    | article_prompt                        # 填入文章生成Prompt
    | llm
    | StrOutputParser()
)

result = article_chain.invoke({"topic": "RAG技术"})

模式3：并行执行多个Chain，然后合并结果

from langchain_core.runnables import RunnableParallel

# 场景：同时让LLM从多个角度分析问题，然后汇总
parallel_chain = RunnableParallel({
    " pros": pros_prompt | llm | StrOutputParser(),   # 分支1：分析优点
    "cons": cons_prompt | llm | StrOutputParser()    # 分支2：分析缺点
})

# parallel_chain.invoke({"topic": "大模型"}) 会返回：
# {"pros": "优点...", "cons": "缺点..."}

# 再接一个汇总Prompt
summary_chain = (
    parallel_chain
    | summary_prompt
    | llm
    | StrOutputParser()
)

---

代码实战6.5：构建多步RAG增强的问答Chain

"""
代码实战6.5：构建带查询改写的多步RAG Chain
问题：用户的原始提问可能表述不清，先改写问题，再做RAG检索
"""

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_core.output_parsers import StrOutputParser
from langchain_core.documents import Document


def build_multi_step_rag_chain():
    """
    构建多步RAG Chain：
    Step 1: 查询改写（让LLM把用户问题改得更适合检索）
    Step 2: 基于改写后的问题做RAG检索
    Step 3: 基于检索结果生成答案
    """

    # 准备测试数据
    docs = [
        Document(page_content="LangChain支持多种大模型接口，包括OpenAI、Anthropic Claude、HuggingFace模型等。通过统一的ChatModel接口，可以用相同的代码调用不同的模型。", metadata={"source": "模型接入"}),
        Document(page_content="RAG（检索增强生成）通过从外部知识库检索相关信息，然后基于检索内容生成答案，有效减少大模型幻觉。RAG的核心步骤包括：文档加载、分块、向量化、检索、生成。", metadata={"source": "RAG原理"}),
        Document(page_content="Agent可以自主规划、调用工具、执行动作。ReAct是一种主流Agent范式，让模型在每一步交替进行思考和行动。AutoGen是微软推出的多Agent协作框架。", metadata={"source": "Agent"})
    ]
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vectorstore = Chroma.from_documents(docs, embeddings, persist_directory="./multi_step_rag_db")
    retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

    llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

    # ========== Step 1：查询改写Chain ==========
    rewrite_prompt = ChatPromptTemplate.from_template("""
原始用户问题：{question}

请把上面的问题改写得更加清晰、完整，使其更适合用于知识库检索。
要求：
1. 保留原问题的核心意图
2. 补充可能缺失的上下文
3. 直接输出改写后的问题，不要有任何解释

改写后的问题：""".strip())

    rewrite_chain = rewrite_prompt | llm | StrOutputParser()

    # ========== Step 2：RAG检索 + 生成 Chain ==========
    rag_prompt = ChatPromptTemplate.from_template("""
请基于以下参考文档回答用户问题。

参考文档：
{context}

用户问题：{question}

回答：""".strip())

    # 完整的多步Chain
    # 技巧：用RunnableLambda包装retriever，使其能在LCEL中使用
    def retrieve_and_format(question: str) -> str:
        docs = retriever.invoke(question)
        return "\n\n".join(doc.page_content for doc in docs)

    full_chain = (
        {
            # "rewritten_question": 先改写问题（这一步可选，演示多步能力）
            "context": RunnableLambda(retrieve_and_format),
            "question": RunnablePassthrough()
        }
        | rag_prompt
        | llm
        | StrOutputParser()
    )

    return full_chain, rewrite_chain, retriever


def demo_multi_step_rag():
    """演示多步RAG Chain"""
    full_chain, rewrite_chain, retriever = build_multi_step_rag_chain()

    # 测试1：直接RAG
    question = "LangChain怎么接入不同的模型？"
    print(f"问题：{question}")
    answer = full_chain.invoke({"question": question})
    print(f"答案：{answer}\n")

    # 测试2：先看查询改写效果
    print("查询改写演示：")
    rewritten = rewrite_chain.invoke({"question": "LangChain咋接各种模型？"})  # 口语化问题
    print(f"原问题：LangChain咋接各种模型？")
    print(f"改写后：{rewritten}")
    print(f"\n用改写后的问题检索：")
    answer2 = full_chain.invoke({"question": rewritten})
    print(f"答案：{answer2}")


if __name__ == "__main__":
    print("=" * 60)
    print("代码实战6.5：多步RAG Chain")
    print("=" * 60)
    demo_multi_step_rag()

---

6.1.5 Agent：构建智能代理

LangChain对Agent的支持是比较早的功能，但API在v0.1之后有较大变化。现在推荐使用 create_tool_calling_agent 来构建Agent。

---

LangChain中Agent的核心组件

┌────────────────────────────────────────┐
│              LangChain Agent         │
│                                    │
│  ┌────────────────────────────┐    │
│  │      Tools（工具列表）      │    │
│  │  - search_tool             │    │
│  │  - calculator_tool         │    │
│  │  - custom_tool            │    │
│  └────────────────────────────┘    │
│              ↓                     │
│  ┌────────────────────────────┐    │
│  │     Agent（决策引擎）      │    │
│  │  （基于LLM + 工具描述）   │    │
│  └────────────────────────────┘    │
│              ↓                     │
│  ┌────────────────────────────┐    │
│  │   AgentExecutor（执行器）  │    │
│  │  （循环执行：思考→行动→观察）│  │
│  └────────────────────────────┘    │
└────────────────────────────────────────┘

---

代码实战6.6：使用LangChain构建支持工具调用的Agent

"""
代码实战6.6：使用LangChain构建支持工具调用的ReAct Agent
完整展示：工具定义 → Agent初始化 → 执行器运行 → 观察结果
"""

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.tools import Tool, BaseTool
from langchain_core.prompts import ChatPromptTemplate
from typing import Dict, Any, Optional
import math


# ========== 方式1：用@tool装饰器定义工具（推荐）==========
from langchain_core.tools import tool

@tool
def calculator(expression: str) -> str:
    """
    计算数学表达式的值。
    支持加减乘除、括号、幂运算（**）。

    Args:
        expression: 数学表达式字符串，如 "2 + 3 * 4" 或 "(10 + 5) / 3"

    Returns:
        计算结果的字符串表示
    """
    try:
        # 安全问题：eval有风险，生产环境应该用ast.literal_eval或专门的表达式解析库
        result = eval(expression, {"__builtins__": {}}, {"math": math})
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算错误：{str(e)}"


@tool
def get_word_length(word: str) -> str:
    """
    计算一个单词的长度。

    Args:
        word: 输入的单词

    Returns:
        单词长度的字符串表示
    """
    return f"{word} 的长度是 {len(word)} 个字符"


# ========== 方式2：继承BaseTool自定义工具（更灵活）==========
class WeatherQueryTool(BaseTool):
    """查询城市天气的工具（模拟）"""

    name = "get_weather"
    description = "查询指定城市的当前天气。当用户询问天气信息时使用此工具。输入应该是城市名称，如'北京'、'上海'。"

    def _run(self, city: str) -> str:
        """工具的实际执行逻辑"""
        # 模拟天气查询（实际应该调用天气API）
        weather_data = {
            "北京": "晴，25°C",
            "上海": "多云，28°C",
            "深圳": "雷阵雨，30°C"
        }
        return weather_data.get(city, f"未找到{city}的天气信息")

    async def _arun(self, city: str) -> str:
        """异步版本（可选实现）"""
        return self._run(city)


def build_agent_with_tools():
    """
    构建支持工具调用的Agent
    """
    # ========== 准备工具列表 ==========
    tools = [calculator, get_word_length, WeatherQueryTool()]

    # ========== 准备LLM ==========
    llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

    # ========== 构建Agent的Prompt ==========
    # 这个Prompt告诉Agent它有哪些工具可用，以及应该如何行动
    prompt = ChatPromptTemplate.from_messages([
        ("system", """你是一个有用的AI助手，可以使用工具来回答用户问题。

当你需要查询信息或进行计算时，使用提供的工具。
如果工具返回了结果，基于结果回答用户问题。
如果没有相关工具或不需要工具就能回答，直接回答。"""),
        ("human", "{input}"),
        ("placeholder", "{agent_scratchpad}")  # Agent的执行轨迹会填在这里
    ])

    # ========== 创建Agent ==========
    # create_tool_calling_agent 是LangChain推荐的新版API
    agent = create_tool_calling_agent(llm, tools, prompt)

    # ========== 创建执行器 ==========
    # AgentExecutor负责循环执行：把Agent的决策变成实际行动
    agent_executor = AgentExecutor(
        agent=agent,
        tools=tools,
        verbose=True,   # 打印执行过程（调试时有用）
        max_iterations=5,  # 最多执行5轮工具调用（防止无限循环）
        handle_parsing_errors=True  # 自动处理LLM输出格式错误
    )

    return agent_executor


def demo_agent_execution(agent_executor):
    """演示Agent的执行过程"""
    test_queries = [
        "帮我说一下北京今天天气怎么样？",
        "计算一下 (10 + 5) * 3 等于多少？",
        "hello 这个单词有多少个字符？"
    ]

    for query in test_queries:
        print("\n" + "=" * 60)
        print(f"用户问题：{query}")
        print("=" * 60)
        result = agent_executor.invoke({"input": query})
        print(f"\n最终答案：{result['output']}")


if __name__ == "__main__":
    print("=" * 60)
    print("代码实战6.6：LangChain工具调用Agent")
    print("=" * 60)

    agent_executor = build_agent_with_tools()
    demo_agent_execution(agent_executor)

运行这段代码时，设置 verbose=True 会打印出Agent的完整思考过程：

> Entering new AgentExecutor chain...

Invoking: `get_weather` with `{'city': '北京'}`
晴，25°C

Invoking: `calculator` with `{'expression': '(10 + 5) * 3'}`
计算结果：45

Final Answer: 北京今天天气晴，25°C。(10 + 5) * 3 = 45

> Finished chain.

这就是ReAct范式在LangChain中的具体体现：思考（决定调用哪个工具）→ 行动（执行工具）→ 观察（拿到工具结果）→ 再思考……直到给出最终答案。

---

6.2 AutoGen多Agent框架实战

6.2.1 AutoGen设计哲学与核心概念

LangChain适合构建"单Agent应用"，但当你需要多个Agent协作时，LangChain的原生支持就比较薄弱了。

微软研究院推出的AutoGen就是专门为多Agent协作设计的框架。它的核心思想是：

把每个Agent看成是一个"对话参与者"，Agent之间通过发消息来协作，就像人在群里聊天一样。

---

AutoGen vs LangChain Agent：核心差异

维度	LangChain Agent	AutoGen
设计焦点	单个Agent的工具调用能力	多个Agent之间的对话协作
Agent定义	基于Tools + Prompt	基于ConversableAgent（可对话的Agent）
协作模式	单一Agent循环执行	多个Agent发消息、形成对话网络
人类介入	支持，但需要自己写逻辑	原生支持人类反馈和介入
适用场景	RAG、单步工具调用	代码生成、复杂任务分解、多角色协作

---

AutoGen的核心抽象

AutoGen里一切都是ConversableAgent——“可以对话的Agent”。每个Agent可以：

• 接收消息
• 思考（调用LLM）
• 发送消息给另一个Agent
• 执行工具（如果配置了function_map）
• 人类介入（如果配置了human_input_mode）

┌────────────────────────────────────────────────┐
│              AutoGen 多Agent系统             │
│                                            │
│  Agent 1 ◄──消息──► Agent 2               │
│    │                                     │
│    │ 消息                                │
│    ▼                                     │
│  Agent 3 ◄──消息──► Human（人类介入点）  │
│                                            │
│  每个Agent都是ConversableAgent的实例        │
│  通过配置决定：是否调用LLM、是否有工具、是否求助人类│
└────────────────────────────────────────────────┘

---

6.2.2 ConversableAgent与群聊管理

ConversableAgent是AutoGen最核心的类。理解了这个类，就理解了AutoGen。

---

ConversableAgent的关键配置参数

from autogen import ConversableAgent

agent = ConversableAgent(
    name="助手",                          # Agent的名字（在对话中显示）
    llm_config={"config_list": [{"model": "gpt-4", "api_key": "..."}]},  # LLM配置
    system_message="你是一个有帮助的助手",  # 系统提示词
    human_input_mode="NEVER",             # 是否需要人类介入（NEVER/ALWAYS/TERMINATE）
    function_map={"calculator": calc_func},  # 注册工具
    code_execution_config=False,           # 是否允许执行代码
    max_consecutive_auto_reply=10,        # 最多连续自动回复多少次（防止无限循环）
)

human_input_mode 有三个选项，这是AutoGen的一个很巧妙的设计：

• "NEVER"：完全自动，不需要人类介入（适合全自动流程）
• "ALWAYS"：每一步都需要人类确认（适合高风险操作）
• "TERMINATE"：只在对话要结束时请求人类确认（平衡点）

---

代码实战6.7：构建双Agent协作系统（代码生成 + 代码审查）

"""
代码实战6.7：使用AutoGen构建双Agent协作系统
场景：Coder Agent写代码 → Reviewer Agent审查代码 → 反馈修改
"""

import autogen
from autogen import ConversableAgent
from typing import Dict, List


def setup_autogen_agents(model_config: Dict):
    """
    初始化两个协作的Agent：
    - Coder：负责写代码
    - Reviewer：负责审查代码，提出修改建议

    Args:
        model_config: LLM配置字典，格式：
            {"config_list": [{"model": "gpt-4", "api_key": "sk-..."}]}
    """

    # ========== Agent 1：Coder（代码编写者）==========
    coder = ConversableAgent(
        name="Coder",
        llm_config=model_config,
        system_message="""你是一个Python代码编写专家。
你的任务：根据用户需求编写高质量Python代码。
要求：
1. 代码要包含详细的中文注释
2. 遵循PEP8规范
3. 考虑边界条件和错误处理
4. 写完代码后，把代码发送给Reviewer审查""",
        human_input_mode="NEVER",       # 不需要人类介入
        code_execution_config=False,     # 不执行代码（只写，不跑）
    )

    # ========== Agent 2：Reviewer（代码审查者）==========
    reviewer = ConversableAgent(
        name="Reviewer",
        llm_config=model_config,
        system_message="""你是一个严格的Python代码审查专家。
你的任务：审查Coder提交的代码，检查：
1. 是否有Bug或边界条件处理不当
2. 注释是否足够清晰
3. 是否有可以优化的地方
4. 是否符合PEP8规范

如果代码质量合格，回复"APPROVED"；
如果发现问题，详细说明问题并建议修改方案，然后把修改建议发回给Coder。""",
        human_input_mode="NEVER",
    )

    return coder, reviewer


def run_coding_workflow(coder, reviewer, task_description: str):
    """
    启动双Agent协作流程

    数据流：
    用户任务 → Coder写代码 → 发给Reviewer审查
                  ↓ (如果APPROVED) 结束
                  ↓ (如果没通过) Reviewer发修改建议 → Coder修改 → 再审查...
    """
    print("\n" + "=" * 60)
    print("启动双Agent协作：代码编写 + 代码审查")
    print("=" * 60)

    # 让Coder开始工作，并把结果发给Reviewer
    # initiate_chat 是AutoGen中启动对话的方法
    result = coder.initiate_chat(
        recipient=reviewer,
        message=f"请帮我写一个Python函数，实现快速排序（QuickSort），要求包含详细中文注释。",
        max_turns=3  # 最多来回3轮（防止无限讨论）
    )

    return result


if __name__ == "__main__":
    print("=" * 60)
    print("代码实战6.7：AutoGen双Agent协作")
    print("=" * 60)

    # 注意：需要设置OPENAI_API_KEY环境变量，或者直接在config中填入
    import os
    model_config = {
        "config_list": [{
            "model": "gpt-3.5-turbo",
            "api_key": os.getenv("OPENAI_API_KEY")
        }]
    }

    if not model_config["config_list"][0]["api_key"]:
        print("请先设置环境变量 OPENAI_API_KEY")
    else:
        coder, reviewer = setup_autogen_agents(model_config)
        run_coding_workflow(coder, reviewer, "实现快速排序")

---

GroupChat：多Agent群聊管理

两个Agent对话用 initiate_chat 就够了，但如果有3个、4个甚至更多Agent呢？

AutoGen提供了 GroupChat 和 GroupChatManager 来管理多Agent群聊：

from autogen import GroupChat, GroupChatManager

# 创建多个Agent
coder = ConversableAgent(name="Coder", ...)
reviewer = ConversableAgent(name="Reviewer", ...)
tester = ConversableAgent(name="Tester", ...)   # 新增：测试者
documenter = ConversableAgent(name="Documenter", ...)  # 新增：文档编写者

# 创建群聊，指定参与者和发言顺序策略
groupchat = GroupChat(
    agents=[coder, reviewer, tester, documenter],
    messages=[],
    max_round=10,  # 最多10轮对话
    # speaker_selection_method: 决定下一轮谁发言
    # "auto"：由LLM自动决定下一发言者（最灵活）
    # "round_robin"：按顺序轮流发言
    speaker_selection_method="auto"
)

# 创建群聊管理器（也是一个Agent，负责协调整个群聊）
manager = GroupChatManager(
    groupchat=groupchat,
    llm_config=model_config
)

# 启动群聊：让Coder发起讨论
coder.initiate_chat(manager, message="我们需要实现一个用户登录功能，请大家协作完成")

GroupChat 的核心价值在于：自动决定下一轮谁发言。用 "speaker_selection_method": "auto" 时，Manager会让LLM根据当前对话内容来判断"接下来谁该说话"，这比固定顺序要灵活得多。

---

6.2.3 人类反馈与干预机制

这是AutoGen相比其他框架的一大优势：人类对Agent的控制是原生支持的，不是事后打补丁。

---

三种人类介入模式详解

模式1："NEVER" — 全自动模式

agent = ConversableAgent(
    name="全自动Agent",
    llm_config=model_config,
    human_input_mode="NEVER"  # 完全不需要人类确认
)
# 适用场景：批量处理、低风险任务、已充分测试的流程

模式2："ALWAYS" — 每步确认模式

agent = ConversableAgent(
    name="谨慎Agent",
    llm_config=model_config,
    human_input_mode="ALWAYS"  # 每轮都请求人类输入
)
# 适用场景：高风险操作（删除文件、发送邮件、金融交易）
# 运行时会暂停，等待人类在终端输入：
#   - 输入内容 → 作为人类反馈发给Agent
#   - 输入 "exit" → 终止对话

模式3："TERMINATE" — 智能介入模式（推荐）

agent = ConversableAgent(
    name="智能介入Agent",
    llm_config=model_config,
    human_input_mode="TERMINATE"  # 只在Agent认为需要人类确认时才介入
)
# Agent在决定"任务完成了"的时候，会请求人类最终确认
# 适用场景：大多数生产环境

---

用人类反馈来纠正Agent的错误

# 场景：Agent生成了代码，但人类发现有个Bug，可以在ALWAYS模式下纠正
# 运行时会提示：
# > Human Input (回车跳过，输入内容发送，'exit'终止):
# 你可以输入：
#   "第三个函数里有个off-by-one错误，请修正"
# Agent会收到这个反馈，然后修改代码

---

6.2.4 AutoGen实战：代码生成与调试Agent

这部分做一个相对完整的实战项目：让多个AutoGen Agent协作完成"需求 → 代码 → 测试 → 调试"的完整流程。

代码实战6.8：完整的代码生成与调试多Agent系统

"""
代码实战6.8：AutoGen多Agent代码生成与调试完整流程
涉及4个Agent角色：
1. ProductManager：把用户需求拆解成技术规格
2. Coder：根据技术规格写代码
3. Tester：写测试用例，发现Bug
4. Debugger：分析测试失败原因，指导Coder修复
"""

import autogen
from autogen import ConversableAgent, GroupChat, GroupChatManager
from typing import Dict


def create_coding_team(model_config: Dict):
    """
    创建一个完整的代码生成与调试团队
    每个Agent有明确的角色定位和退出条件
    """

    # ========== Agent 1：产品经理（需求分析）==========
    pm = ConversableAgent(
        name="ProductManager",
        llm_config=model_config,
        system_message="""你是一个产品经理AI。
你的输入：用户的模糊需求（如"帮我做个待办事项应用"）
你的输出：清晰的技术规格说明，包括：
1. 核心功能列表
2. 每个功能的输入输出
3. 边界条件和异常处理要求
4. 建议的代码结构

输出格式要清晰，方便Coder直接基于此写代码。""",
        human_input_mode="NEVER",
    )

    # ========== Agent 2：程序员（代码编写）==========
    coder = ConversableAgent(
        name="Coder",
        llm_config=model_config,
        system_message="""你是一个Python代码编写专家。
你的任务：根据ProductManager提供的技术规格，编写完整的Python代码。
要求：
1. 包含详细的中文注释
2. 遵循PEP8规范
3. 包含所有必要的异常处理
4. 代码要可以直接运行

你写完代码后，把代码发给Tester进行测试。""",
        human_input_mode="NEVER",
        code_execution_config={
            "work_dir": "autogen_workspace",  # 代码执行的工作目录
            "use_docker": False  # 不用Docker（生产环境建议用Docker沙箱）
        }  # 允许执行代码！
    )

    # ========== Agent 3：测试工程师（写测试、运行测试）==========
    tester = ConversableAgent(
        name="Tester",
        llm_config=model_config,
        system_message="""你是一个严格的测试工程师AI。
你的任务：
1. 根据技术规格，为Coder的代码编写完整的测试用例
2. 运行测试用例（你可以执行代码）
3. 如果测试全部通过，回复"ALL_TESTS_PASSED"
4. 如果有测试失败，详细描述了失败原因，并把失败信息发给Debugger

你生成的测试代码要包含：
- 正常情况的测试
- 边界情况的测试
- 异常情况的测试""",
        human_input_mode="NEVER",
        code_execution_config={
            "work_dir": "autogen_workspace",
            "use_docker": False
        }
    )

    # ========== Agent 4：调试工程师（分析错误、指导修复）==========
    debugger = ConversableAgent(
        name="Debugger",
        llm_config=model_config,
        system_message="""你是一个调试专家AI。
你的任务：
1. 分析Tester报告的测试失败信息
2. 定位代码中的Bug原因
3. 给出具体的修复建议（要详细到Coder可以直接改）
4. 把修复建议发给Coder

如果经过3轮调试仍然无法修复，回复"DEBUG_FAILED: [原因]"，终止流程。""",
        human_input_mode="NEVER",
    )

    return pm, coder, tester, debugger


def run_full_coding_workflow(model_config: Dict, user_requirement: str):
    """
    运行完整的代码生成与调试流程
    数据流：
    用户需求 → PM分析 → Coder写代码 → Tester测试
      ↓ (测试失败) → Debugger分析 → Coder修改 → Tester再测试
      ↓ (测试通过) → 输出最终代码
    """

    pm, coder, tester, debugger = create_coding_team(model_config)

    # 创建GroupChat，让4个Agent协作
    groupchat = GroupChat(
        agents=[pm, coder, tester, debugger],
        messages=[],
        max_round=15,  # 最多15轮对话
        speaker_selection_method="auto"  # LLM自动选择下一发言者
    )

    manager = GroupChatManager(
        groupchat=groupchat,
        llm_config=model_config
    )

    print("\n" + "=" * 60)
    print("启动完整的代码生成与调试多Agent系统")
    print(f"用户需求：{user_requirement}")
    print("=" * 60)

    # PM发起讨论
    result = pm.initiate_chat(
        manager,
        message=f"新需求来了：{user_requirement}\n请先分析需求，输出技术规格。",
        max_turns=15
    )

    return result


if __name__ == "__main__":
    print("=" * 60)
    print("代码实战6.8：AutoGen完整代码生成与调试系统")
    print("=" * 60)

    import os
    model_config = {
        "config_list": [{
            "model": "gpt-3.5-turbo",
            "api_key": os.getenv("OPENAI_API_KEY")
        }]
    }

    if not model_config["config_list"][0]["api_key"]:
        print("请先设置环境变量 OPENAI_API_KEY")
    else:
        # 运行完整流程
        result = run_full_coding_workflow(
            model_config,
            user_requirement="写一个Python函数，实现二分查找算法，并在有序数组中查找目标元素"
        )
        print("\n最终对话历史：")
        for msg in result.chat_history:
            print(f"[{msg['role']}] {msg['content'][:100]}...")

---

6.3 框架选型与对比

6.3.1 LangChain vs AutoGen：适用场景对比

这两个框架经常被拿来比较，但它们的定位其实不太一样。

维度	LangChain	AutoGen
核心定位	LLM应用开发的全能框架	多Agent协作的专业框架
最佳场景	RAG应用、单Agent工具调用、Prompt工程	多角色协作任务、代码生成、复杂问题分解
学习曲线	中等（概念多，API演变快）	较平缓（核心概念少）
工具调用	非常完善（Tool、Toolkit、AgentExecutor）	支持，但不是核心焦点
多Agent协作	基础支持（需要自己编排）	核心能力（GroupChat原生支持）
人类介入	需要自己实现	原生支持（human_input_mode）
代码执行	需要自己集成	原生支持（code_execution_config）
社区&生态	非常庞大（最多人用）	中等（微软背书，增长快）
生产稳定性	较好（很多公司在用）	较好（微软内部在用）

选型建议：

• 做RAG应用 → 选LangChain（Retrieval模块非常成熟）
• 做单Agent工具调用 → LangChain和AutoGen都可以，LangChain的工具定义更规范
• 做多Agent协作（如代码生成团队） → 选AutoGen（GroupChat是杀手级功能）
• 两个都用 → 实际上可以混合使用：用LangChain做RAG检索，把结果送给AutoGen的Agent处理

---

6.3.2 其他框架概览

除了LangChain和AutoGen，还有几个值得关注的框架：

CrewAI：角色驱动的多Agent框架

CrewAI的设计灵感来自"一个团队（Crew）里有不同角色的人，各自有专长，协作完成任务"。

# CrewAI的核心抽象
from crewai import Agent, Task, Crew

# 定义Agent（指定角色、目标、工具）
researcher = Agent(
    role="高级研究员",
    goal="收集关于某个技术的最新信息",
    backstory="你是一个经验丰富的研究员，擅长信息检索和分析",
    tools=[search_tool]
)

writer = Agent(
    role="技术作家",
    goal="把研究结果写成易懂的文章",
    backstory="你是一个优秀的技术作家，擅长把复杂概念讲清楚"
)

# 定义Task（任务）
research_task = Task(
    description="研究RAG技术的最新进展",
    agent=researcher
)

write_task = Task(
    description="基于研究结果写一篇科普文章",
    agent=writer
)

# 组建Crew（团队），指定执行策略
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process="sequential"  # 顺序执行：先研究，再写文章
)

result = crew.kickoff()

CrewAI vs AutoGen：

• CrewAI更适合"角色职责明确、按顺序执行"的场景
• AutoGen更适合"需要灵活对话、动态决定下一发言人"的场景

---

Semantic Kernel：微软的另一个Agent框架

Semantic Kernel（SK）是微软推出的另一款框架，定位更偏向企业.NET/Java/Python全栈，和AutoGen是"同门师兄弟"。

# Semantic Kernel的核心抽象：Kernel + Plugin + Planner
import semantic_kernel as sk
from semantic_kernel.connectors.ai.open_ai import OpenAIChatCompletion

kernel = sk.Kernel()
kernel.add_service(OpenAIChatCompletion(model_id="gpt-4", api_key="..."))

# Plugin = 工具集合（类似LangChain的Toolkit）
# Planner = 自动规划器（类似Agent的规划能力）

Semantic Kernel的特点：

• 对.NET/C#支持最好（如果用C#写后端，SK是首选）
• Plugin系统很规范（比LangChain的Tool定义更结构化）
• Planner功能比LangChain的Agent弱一些

---

Haystack：专注于RAG和搜索的框架

Haystack是deepset.ai推出的框架，专注在搜索+RAG场景，在NLP学术界用得比较多。

如果你要做的东西很接近"问答系统"、“文档检索”，Haystack可能是比LangChain更专业的选择。

---

6.3.3 自定义框架设计考虑

有时候，现成框架不能满足你的需求。比如：

• 需要极致的推理速度（框架的抽象层带来了额外延迟）
• 需要接入非常定制化的工具或数据源
• 需要在资源受限的环境运行（框架本身太重）

什么情况下应该自研框架？

# 一个极简的"自研Agent框架"核心循环，不到50行代码
def minimal_agent_loop(
    llm_call,           # 函数：调用LLM
    tools: dict,         # 字典：工具名 → 工具函数
    system_prompt: str,
    max_iterations: int = 10
):
    """
    极简Agent循环：思考 → 行动 → 观察
    适合对性能和可控性要求极高的场景
    """
    messages = [{"role": "system", "content": system_prompt}]

    for i in range(max_iterations):
        # === 思考：让LLM决定下一步 ===
        response = llm_call(messages)

        if response.finish_reason == "stop":
            # LLM认为任务完成了，直接返回最终答案
            return response.content

        # === 行动：解析工具调用 ===
        tool_call = parse_tool_call(response)  # 解析LLM输出中的工具调用
        tool_name = tool_call["name"]
        tool_params = tool_call["parameters"]

        # === 执行：调用实际工具 ===
        if tool_name in tools:
            observation = tools[tool_name](**tool_params)
        else:
            observation = f"错误：工具 {tool_name} 不存在"

        # === 观察：把结果加回消息历史 ===
        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": f"工具执行结果：{observation}"})

    return "达到最大迭代次数，任务可能未完成。"

自研框架的优劣势：

优势	劣势
极致性能（无框架开销）	开发成本高，需要自己处理各种边界情况
完全可控（每个细节都能改）	维护成本高，框架团队的工作你得自己做
可以深度优化特定场景	生态为零，没有社区插件可用
	新员工需要学习你的自研框架（学习成本）

建议：

• 90%的场景用现成框架（LangChain/AutoGen）
• 只有当你对性能、可控性有极端要求时，才考虑自研
• 如果自研，参考LangChain的Runnable接口设计和AutoGen的ConversableAgent设计，不要从头造轮子

---

6.3.4 框架性能与可扩展性评估

选框架时，性能是一个重要考量。但这里说的"性能"不止是"跑得快"，还包括：

┌─────────────────────────────────────────────┐
│           框架性能评估维度                 │
│                                         │
│  1. 推理延迟：从发请求到拿到完整回复的时间 │
│  2. 吞吐量：单位时间内能处理多少请求       │
│  3. 内存占用：框架本身占多少内存          │
│  4. 可扩展性：能否方便地接入新工具/新模型 │
│  5. 并发能力：能否同时处理多个用户请求    │
│  6. 部署便利性：是否容易容器化、微服务化  │
└─────────────────────────────────────────────┘

---

推理延迟对比（定性分析）

直接调用API（无框架）：        基准线（100%）
LangChain LCEL：               +5~15%（主要是Prompt组装和输出解析的开销）
LangChain AgentExecutor：       +20~40%（需要多轮循环，每轮都有Prompt组装开销）
AutoGen ConversableAgent：    +15~30%（主要是消息历史管理开销）
自研极简循环：               +5~10%

结论：框架带来的额外延迟通常不超过50ms，相比LLM本身的推理时间（几百ms到几秒），这个开销可以忽略。不要为了"性能"而牺牲框架的开发效率。

---

可扩展性：接入新模型/新工具有多容易？

# LangChain：接入新工具（非常简单）
@tool
def my_custom_tool(param: str) -> str:
    """工具描述，LLM会看到这个。"""
    return do_something(param)

# LangChain：接入新模型（也比较简单）
from langchain_core.language_models import BaseChatModel

class MyCustomChatModel(BaseChatModel):
    """只需要实现_generate方法，就能接入任何自定义模型"""
    def _generate(self, messages, stop=None, run_manager=None, **kwargs):
        # 调用你的自定义模型API
        response = my_custom_api(messages)
        return AIMessage(content=response)

---

企业级避坑指南

避坑1：LangChain版本升级导致代码大面积报错

问题描述：

LangChain在2023-2024年间经历了两次较大的API变更（v0.0 → v0.1 LCEL，v0.1 → v0.2），网上的很多教程代码是基于旧版API的，直接抄来用会报错。

典型报错：

TypeError: LLMChain.__init__() got an unexpected keyword argument 'llm'
# 这是旧版API的写法，新版已经不支持

解决方案：

1. 固定版本：在requirements.txt中固定LangChain版本

langchain==0.2.0
langchain-openai==0.1.0
langchain-community==0.2.0

2. 使用现代API：所有新代码都用LCEL（prompt | llm | output_parser），不要用旧的LLMChain、ConversationChain等类

3. 迁移旧代码：参考官方迁移指南

# 旧版（不要用）
from langchain.chains import LLMChain
chain = LLMChain(llm=llm, prompt=prompt)
result = chain.run(input="...")

# 新版（推荐）
chain = prompt | llm | StrOutputParser()
result = chain.invoke({"input": "..."})

---

避坑2：AutoGen的GroupChat陷入无限对话

问题描述：

配置了GroupChat后，多个Agent来回对话，但没有一个Agent"喊停"，导致对话无限继续，API费用快速消耗。

解决方案：

groupchat = GroupChat(
    agents=[agent1, agent2, agent3],
    messages=[],
    max_round=10,  # ← 一定要设置这个！限制最多多少轮对话
    speaker_selection_method="auto"
)

# 另外，在每个Agent的system_message里明确"退出条件"：
system_message="""...
如果任务已完成，请明确说"TASK_COMPLETED"，这样GroupChat就会结束。
不要在没有新信息的情况下重复之前的发言。"""

---

避坑3：框架抽象泄漏，调试困难

问题描述：

用框架写代码很爽，但一旦出问题（比如Agent不调用工具、RAG检索结果不对），很难定位是"框架的问题"还是"自己代码的问题"，因为框架封装了很多细节。

解决方案：

1. 开启详细日志：

import logging

# LangChain日志
logging.basicConfig(level=logging.DEBUG)

# AutoGen日志
autogen.logging_config = {
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {"default": {"format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s"}},
    "handlers": {"default": {"class": "logging.StreamHandler", "formatter": "default"}},
    "loggers": {"autogen": {"handlers": ["default"], "level": "DEBUG"}}
}

2. 先不用框架写一遍原型，确认逻辑正确后，再用框架重构。这样你清楚"框架在帮你做什么"，出问题时能快速定位。

3. 为关键步骤写单元测试：

# 测试RAG检索是否正常工作（不依赖LLM）
def test_retriever():
    retriever = build_retriever()
    docs = retriever.invoke("测试查询")
    assert len(docs) > 0, "检索器没有返回任何文档"

---

本章小结

核心Takeaways

1. LangChain的核心价值在于把LLM应用开发的通用模式抽象成标准组件（Model I/O、Retrieval、Chain、Agent、Memory），用LCEL可以把这些组件像管道一样串起来。

2. LCEL（LangChain Expression Language） 是现代LangChain的核心，用 prompt | llm | output_parser 的管道语法替代了旧的LLMChain类，代码更清晰、更灵活。

3. RAG在LangChain中的完整流程：Document Loader → Text Splitter → Embeddings → VectorStore → Retriever → 填入Prompt → LLM生成答案，每一步都有对应的标准组件。

4. AutoGen的核心设计哲学是"Agent之间的对话即协作"，ConversableAgent可以发消息、调用工具、执行代码、请求人类介入，非常适合多Agent协作场景。

5. AutoGen的GroupChat 支持多个Agent自动协作，通过 speaker_selection_method="auto"让LLM动态决定下一轮谁发言，比固定顺序的协作模式灵活得多。

6. 框架选型建议：RAG应用选LangChain，多Agent协作选AutoGen，两个可以混合使用。90%的场景不需要自研框架。

---

思考题

思考题1：

你在用LangChain构建一个RAG应用时，发现检索到的文档块和问题"语义相关但实际没用"（比如问题问"如何使用API"，检索到了"API的架构设计"，内容相关但不是用户想要的）。请设计一种检索后处理方案来提升最终答案质量，可以用到本章介绍的哪些LangChain组件？

参考答案要点

核心思路：检索只是第一步，需要在检索之后、送给LLM之前，增加一个"重排序（Reranking）"步骤。

具体方案：

1. 用LangChain的Retriever先拉取Top-20（多拿一些）
2. 用一个轻量的Reranker模型（如Cohere Rerank、或BGE Reranker）对这20个文档块重新打分排序
3. 取Reranker排序后的Top-5送给LLM

用到的LangChain组件：

• 可以自定义一个BaseRetriever的子类，把Reranker逻辑封装进去
• 或者用LCEL： retriever | reranker | format_docs | prompt | llm
• LangChain社区版也提供了ContextualCompressionRetriever，就是做这件事的

延伸：还可以结合"查询改写"（用LLM把用户问题改写成更适合检索的版本），进一步提升检索精度。

---

思考题2：

AutoGen的GroupChat中，如果有一个Agent"太能说"（比如Coder Agent每次都输出很长的代码解释，导致对话历史很快超出LLM的上下文窗口），请设计两种不同的策略来解决这个问题，并分析各自的优缺点。

参考答案要点

策略1：限制每个Agent的单次输出长度

• 在Agent的system_message里明确指示：“每次回复控制在200字以内，代码用代码块展示”
• 优点：简单，不需要改框架代码
• 缺点：LLM不一定遵守，需要加输出解析器强制截断

策略2：用max_consecutive_auto_reply限制连续发言次数

coder = ConversableAgent(
    name="Coder",
    llm_config=model_config,
    max_consecutive_auto_reply=2  # ← 最多连续说2次话
)

• 优点：框架原生支持，效果好
• 缺点：可能打断Agent的合理长回复（比如确实需要从头写到尾的代码）

策略3（推荐）：启用code_execution_config，让代码在沙箱里执行，对话中只传"执行结果摘要"

• 优点：代码不占对话历史，只传执行结果
• 缺点：需要配置沙箱环境（Docker），部署复杂度增加

策略4：定期Summary（用另一个Agent负责总结对话历史，压缩上下文）

• AutoGen社区有这个功能的讨论，但尚未成熟，可以自己实现

---