引言

在Agent开发领域，LangChain无疑是最受欢迎的框架之一。它提供了一套完整的工具链，帮助开发者快速构建强大的AI应用。LangChain的核心优势在于它的模块化设计和丰富的集成能力，能够轻松连接各种LLM、数据源和外部工具。

本章将详细介绍LangChain的核心概念，包括Chain、Prompt、Tool等，并通过实战案例演示如何使用LangChain构建Agent。

一、LangChain概述

1.1 什么是LangChain

LangChain是一个用于构建基于LLM的应用程序的框架。它提供了一套工具和抽象，使得开发者能够快速构建复杂的AI应用，包括聊天机器人、问答系统、代码生成器等。

1.2 LangChain的核心优势

优势	说明
模块化设计	将复杂的AI应用分解为可组合的模块
丰富的集成	支持多种LLM、向量数据库、API等
易于扩展	提供插件系统，方便添加自定义功能
文档完善	提供详细的文档和丰富的示例
社区活跃	拥有庞大的开发者社区和生态系统

1.3 LangChain的架构

┌─────────────────────────────────────────────────────────────┐
│                        LangChain                            │
├─────────────────────────────────────────────────────────────┤
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐       │
│  │   LLM层      │  │   工具层     │  │   数据层     │       │
│  │  OpenAI      │  │  Search      │  │  Vector DB   │       │
│  │  Anthropic   │  │  Calculator  │  │  SQL DB      │       │
│  │  Google      │  │  API调用     │  │  文件系统    │       │
│  └──────┬───────┘  └──────┬───────┘  └──────┬───────┘       │
│         │                 │                 │                │
│         └────────┬────────┴────────┬────────┘                │
│                  │                 │                         │
│         ┌────────▼────────┐  ┌─────▼─────┐                  │
│         │    Chain层      │  │ Memory层  │                  │
│         │  SequentialChain│  │  Buffer   │                  │
│         │  RouterChain   │  │  Summary  │                  │
│         │  TransformChain│  │  Vector   │                  │
│         └──────┬─────────┘  └────┬──────┘                  │
│                │                 │                          │
│                └────────┬────────┘                          │
│                         ▼                                   │
│              ┌─────────────────┐                            │
│              │    Agent层      │                            │
│              │  ReAct Agent    │                            │
│              │  Conversational │                            │
│              │  PlanAndExecute │                            │
│              └─────────────────┘                            │
└─────────────────────────────────────────────────────────────┘

1.4 安装LangChain

# 安装基础版本
pip install langchain

# 安装完整版本（包含所有依赖）
pip install langchain[all]

# 安装特定模块
pip install langchain[openai]  # OpenAI集成
pip install langchain[chromadb]  # Chroma向量数据库
pip install langchain[anthropic]  # Anthropic集成

---

二、核心概念：Prompt

2.1 Prompt的定义

Prompt是指用户输入给LLM的文本，用于引导LLM生成特定的输出。一个好的Prompt能够让LLM更准确地理解用户意图，生成高质量的响应。

2.2 Prompt模板

LangChain提供了Prompt模板功能，允许开发者定义可复用的Prompt结构。

2.2.1 基础Prompt模板

from langchain.prompts import PromptTemplate

# 创建基础Prompt模板
prompt = PromptTemplate(
    input_variables=["topic"],
    template="请介绍一下{topic}。"
)

# 使用模板
result = prompt.format(topic="人工智能")
print(result)

2.2.2 ChatPrompt模板

from langchain.prompts import ChatPromptTemplate

# 创建ChatPrompt模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一名专业的技术顾问。"),
    ("human", "请介绍一下{topic}。")
])

# 使用模板
result = prompt.format(topic="LangChain")
print(result)

2.2.3 带示例的Prompt模板

from langchain.prompts import FewShotPromptTemplate, PromptTemplate

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

# 创建示例
examples = [
    {"question": "什么是Python?", "answer": "Python是一种高级编程语言。"},
    {"question": "什么是机器学习?", "answer": "机器学习是一种人工智能技术。"},
    {"question": "什么是深度学习?", "answer": "深度学习是机器学习的一个分支。"}
]

# 创建FewShotPrompt模板
prompt = FewShotPromptTemplate(
    examples=examples,
    example_prompt=example_prompt,
    prefix="请根据以下示例回答问题:",
    suffix="问题: {input}\n答案:",
    input_variables=["input"]
)

# 使用模板
result = prompt.format(input="什么是LangChain?")
print(result)

2.3 Prompt最佳实践

2.3.1 明确任务描述

from langchain.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一名专业的代码审查员。请仔细审查以下代码，并提供详细的改进建议。

审查标准：
1. 代码可读性：命名规范、注释清晰
2. 代码质量：避免重复、遵循DRY原则
3. 安全性：防止注入攻击、敏感信息保护
4. 性能：避免不必要的计算、优化算法

输出格式：
## 问题列表
1. [问题描述] - [严重程度] - [改进建议]

## 优化建议
1. [建议内容]"""),
    ("human", "{code}")
])

2.3.2 使用结构化输出

from langchain.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", """请分析以下文本，并以JSON格式输出结果。

输出格式：
{{
    "sentiment": "positive|negative|neutral",
    "entities": [{"name": "实体名称", "type": "实体类型"}],
    "summary": "文本摘要"
}}"""),
    ("human", "{text}")
])

2.3.3 限制输出长度

from langchain.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", """请回答以下问题，限制在100字以内。"""),
    ("human", "{question}")
])

---

三、核心概念：Chain

3.1 Chain的定义

Chain是LangChain中最核心的概念之一。它是一系列组件的组合，用于处理输入并生成输出。Chain可以将多个组件串联起来，形成一个完整的处理流程。

3.2 基础Chain

3.2.1 LLMChain

from langchain.chat_models import ChatOpenAI
from langchain.chains import LLMChain
from langchain.prompts import ChatPromptTemplate

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

# 创建Prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一名助手。"),
    ("human", "{input}")
])

# 创建Chain
chain = LLMChain(llm=llm, prompt=prompt)

# 运行Chain
result = chain.run(input="请介绍一下LangChain。")
print(result)

3.2.2 异步Chain

# 异步运行Chain
result = await chain.arun(input="请介绍一下LangChain。")
print(result)

3.3 组合Chain

3.3.1 SequentialChain

SequentialChain将多个Chain按顺序连接起来，前一个Chain的输出作为后一个Chain的输入。

from langchain.chat_models import ChatOpenAI
from langchain.chains import LLMChain, SequentialChain
from langchain.prompts import ChatPromptTemplate

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

# 创建第一个Chain：生成主题
prompt1 = ChatPromptTemplate.from_messages([
    ("system", "你是一名创意写作助手。"),
    ("human", "请为一篇技术文章生成一个主题。")
])
chain1 = LLMChain(llm=llm, prompt=prompt1, output_key="topic")

# 创建第二个Chain：生成大纲
prompt2 = ChatPromptTemplate.from_messages([
    ("system", "你是一名技术编辑。"),
    ("human", "请为主题'{topic}'生成文章大纲。")
])
chain2 = LLMChain(llm=llm, prompt=prompt2, output_key="outline")

# 创建第三个Chain：生成文章
prompt3 = ChatPromptTemplate.from_messages([
    ("system", "你是一名技术作家。"),
    ("human", "请根据大纲'{outline}'写一篇详细的文章。")
])
chain3 = LLMChain(llm=llm, prompt=prompt3, output_key="article")

# 组合Chain
overall_chain = SequentialChain(
    chains=[chain1, chain2, chain3],
    input_variables=[],
    output_variables=["topic", "outline", "article"],
    verbose=True
)

# 运行组合Chain
result = overall_chain.run({})
print(f"主题: {result['topic']}")
print(f"大纲: {result['outline']}")
print(f"文章: {result['article']}")

3.3.2 RouterChain

RouterChain根据输入内容选择不同的Chain进行处理。

from langchain.chat_models import ChatOpenAI
from langchain.chains import LLMChain, RouterChain, MultiPromptChain
from langchain.prompts import ChatPromptTemplate
from langchain.chains.router.llm_router import LLMRouterChain, RouterOutputParser
from langchain.chains.router.multi_prompt_prompt import MULTI_PROMPT_ROUTER_TEMPLATE

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

# 创建不同领域的Prompt
physics_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一名物理学专家。"),
    ("human", "{input}")
])

math_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一名数学专家。"),
    ("human", "{input}")
])

computer_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一名计算机科学专家。"),
    ("human", "{input}")
])

# 创建Prompt字典
prompt_infos = [
    {
        "name": "physics",
        "description": "处理物理学相关问题",
        "prompt_template": physics_prompt
    },
    {
        "name": "math",
        "description": "处理数学相关问题",
        "prompt_template": math_prompt
    },
    {
        "name": "computer",
        "description": "处理计算机科学相关问题",
        "prompt_template": computer_prompt
    }
]

# 创建RouterChain
destination_chains = {}
for p_info in prompt_infos:
    name = p_info["name"]
    prompt = p_info["prompt_template"]
    chain = LLMChain(llm=llm, prompt=prompt)
    destination_chains[name] = chain

# 创建路由Prompt
router_template = MULTI_PROMPT_ROUTER_TEMPLATE.format(
    destinations="\n".join([f"{p['name']}: {p['description']}" for p in prompt_infos])
)
router_prompt = ChatPromptTemplate.from_messages([
    ("system", router_template),
    ("human", "{input}")
])

# 创建RouterChain
router_chain = LLMRouterChain.from_llm(llm, router_prompt)

# 创建MultiPromptChain
chain = MultiPromptChain(
    router_chain=router_chain,
    destination_chains=destination_chains,
    default_chain=destination_chains["computer"]
)

# 运行Chain
result = chain.run("什么是量子力学？")
print(result)

result = chain.run("什么是深度学习？")
print(result)

3.3.3 TransformChain

TransformChain允许开发者自定义输入输出转换逻辑。

from langchain.chains import TransformChain
from langchain.chains import SequentialChain

# 创建TransformChain
def transform_func(inputs):
    text = inputs["text"]
    # 转换逻辑：将文本转换为大写
    transformed_text = text.upper()
    return {"transformed_text": transformed_text}

transform_chain = TransformChain(
    input_variables=["text"],
    output_variables=["transformed_text"],
    transform=transform_func
)

# 创建LLMChain
from langchain.chat_models import ChatOpenAI
from langchain.chains import LLMChain
from langchain.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一名助手。"),
    ("human", "请分析以下文本：{transformed_text}")
])
llm_chain = LLMChain(llm=llm, prompt=prompt, output_key="analysis")

# 组合Chain
overall_chain = SequentialChain(
    chains=[transform_chain, llm_chain],
    input_variables=["text"],
    output_variables=["transformed_text", "analysis"],
    verbose=True
)

# 运行Chain
result = overall_chain.run(text="Hello, LangChain!")
print(f"转换后的文本: {result['transformed_text']}")
print(f"分析结果: {result['analysis']}")

---

四、核心概念：Tool

4.1 Tool的定义

Tool是Agent与外部世界交互的桥梁。通过Tool，Agent可以执行各种操作，如搜索、计算、文件操作等。

4.2 内置工具

LangChain提供了许多内置工具，可以直接使用。

4.2.1 搜索工具

from langchain.tools import DuckDuckGoSearchRun

# 创建搜索工具
search = DuckDuckGoSearchRun()

# 使用搜索工具
result = search.run("LangChain最新版本")
print(result)

4.2.2 计算器工具

from langchain.tools import CalculatorTool

# 创建计算器工具
calculator = CalculatorTool()

# 使用计算器工具
result = calculator.run("25 × 48 + 120")
print(result)

4.2.3 文件操作工具

from langchain.tools import FileWriteTool, FileReadTool

# 创建文件写入工具
write_tool = FileWriteTool()

# 使用文件写入工具
result = write_tool.run({"file_path": "output.txt", "content": "Hello, LangChain!"})
print(result)

# 创建文件读取工具
read_tool = FileReadTool()

# 使用文件读取工具
result = read_tool.run("output.txt")
print(result)

4.3 自定义工具

开发者可以根据需要创建自定义工具。

4.3.1 基础自定义工具

from langchain.tools import BaseTool
from typing import Optional, Type
from pydantic import BaseModel, Field

class SearchInput(BaseModel):
    query: str = Field(description="搜索查询")

class CustomSearchTool(BaseTool):
    name = "custom_search"
    description = "使用自定义搜索引擎搜索信息"
    args_schema: Type[BaseModel] = SearchInput
    
    def _run(self, query: str) -> str:
        # 自定义搜索逻辑
        return f"搜索结果: {query} - 这是模拟的搜索结果"
    
    async def _arun(self, query: str) -> str:
        # 异步搜索逻辑
        return f"搜索结果: {query} - 这是模拟的异步搜索结果"

# 使用自定义工具
tool = CustomSearchTool()
result = tool.run("LangChain教程")
print(result)

4.3.2 带参数验证的工具

from langchain.tools import BaseTool
from typing import Optional, Type
from pydantic import BaseModel, Field, ValidationError

class CalculatorInput(BaseModel):
    expression: str = Field(description="数学表达式")
    
    @classmethod
    def validate_expression(cls, v):
        # 验证表达式只包含数字和运算符
        allowed_chars = "0123456789+-*/(). "
        for char in v:
            if char not in allowed_chars:
                raise ValueError(f"表达式包含非法字符: {char}")
        return v

class SafeCalculatorTool(BaseTool):
    name = "safe_calculator"
    description = "安全计算器，只接受合法的数学表达式"
    args_schema: Type[BaseModel] = CalculatorInput
    
    def _run(self, expression: str) -> str:
        try:
            # 安全计算
            result = eval(expression)
            return f"计算结果: {result}"
        except Exception as e:
            return f"计算错误: {e}"

# 使用工具
tool = SafeCalculatorTool()

# 合法表达式
result = tool.run("25 * 48")
print(result)

# 非法表达式（包含注入攻击）
result = tool.run("__import__('os').system('ls')")
print(result)

4.4 工具集成

4.4.1 集成API

from langchain.tools import BaseTool
from typing import Optional, Type
from pydantic import BaseModel, Field
import requests

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
        url = f"https://api.openweathermap.org/data/2.5/weather?q={city}&appid=your-api-key"
        response = requests.get(url)
        data = response.json()
        
        if response.status_code == 200:
            weather = data["weather"][0]["description"]
            temp = data["main"]["temp"] - 273.15
            return f"{city}的天气: {weather}, 温度: {temp:.1f}°C"
        else:
            return f"查询失败: {data.get('message', '未知错误')}"

# 使用工具
tool = WeatherTool()
result = tool.run("Beijing")
print(result)

4.4.2 集成数据库

from langchain.tools import BaseTool
from typing import Optional, Type
from pydantic import BaseModel, Field
import sqlite3

class DatabaseQueryInput(BaseModel):
    query: str = Field(description="SQL查询语句")

class DatabaseTool(BaseTool):
    name = "database_query"
    description = "查询SQLite数据库"
    args_schema: Type[BaseModel] = DatabaseQueryInput
    
    def __init__(self, db_path: str):
        super().__init__()
        self.db_path = db_path
    
    def _run(self, query: str) -> str:
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            
            # 限制只能执行SELECT查询
            if not query.strip().upper().startswith("SELECT"):
                return "只允许执行SELECT查询"
            
            cursor.execute(query)
            results = cursor.fetchall()
            
            conn.close()
            
            return f"查询结果: {results}"
        except Exception as e:
            return f"查询错误: {e}"

# 使用工具
tool = DatabaseTool(db_path="example.db")
result = tool.run("SELECT * FROM users")
print(result)

---

五、核心概念：Memory

5.1 Memory的定义

Memory允许Agent在对话过程中记住之前的交互内容，从而实现上下文理解和多轮对话。

5.2 内置Memory类型

5.2.1 ConversationBufferMemory

from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
from langchain.chat_models import ChatOpenAI

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

# 创建Memory
memory = ConversationBufferMemory()

# 创建对话Chain
conversation = ConversationChain(llm=llm, memory=memory, verbose=True)

# 进行对话
response1 = conversation.predict(input="你好，我叫张三。")
print(response1)

response2 = conversation.predict(input="我是一名程序员。")
print(response2)

response3 = conversation.predict(input="我擅长Python编程。")
print(response3)

response4 = conversation.predict(input="你知道我叫什么名字吗？")
print(response4)

5.2.2 ConversationBufferWindowMemory

from langchain.memory import ConversationBufferWindowMemory

# 创建带窗口的Memory，只保留最近2轮对话
memory = ConversationBufferWindowMemory(k=2)

# 创建对话Chain
conversation = ConversationChain(llm=llm, memory=memory, verbose=True)

# 进行多轮对话
conversation.predict(input="你好，我叫张三。")
conversation.predict(input="我是一名程序员。")
conversation.predict(input="我擅长Python编程。")
conversation.predict(input="我喜欢打篮球。")

# 由于窗口大小为2，只会记住最近2轮对话
response = conversation.predict(input="你知道我叫什么名字吗？")
print(response)

5.2.3 ConversationSummaryMemory

from langchain.memory import ConversationSummaryMemory

# 创建带摘要的Memory
memory = ConversationSummaryMemory(llm=llm)

# 创建对话Chain
conversation = ConversationChain(llm=llm, memory=memory, verbose=True)

# 进行多轮对话
conversation.predict(input="你好，我叫张三。")
conversation.predict(input="我是一名程序员，在一家科技公司工作。")
conversation.predict(input="我擅长Python和Java编程，已经有5年工作经验了。")
conversation.predict(input="我喜欢打篮球和看电影。")

# 查看摘要
print(memory.buffer)

5.2.4 ConversationKGMemory

from langchain.memory import ConversationKGMemory

# 创建知识图谱Memory
memory = ConversationKGMemory(llm=llm)

# 创建对话Chain
conversation = ConversationChain(llm=llm, memory=memory, verbose=True)

# 进行对话
conversation.predict(input="张三是一名程序员，他在阿里巴巴工作。")
conversation.predict(input="李四是一名设计师，她在腾讯工作。")

# 查看知识图谱
print(memory.kg)

5.3 自定义Memory

from langchain.memory import BaseMemory
from typing import Dict, List
from pydantic import BaseModel

class CustomMemory(BaseMemory, BaseModel):
    history: List[Dict[str, str]] = []
    max_length: int = 10
    
    @property
    def memory_variables(self) -> List[str]:
        return ["history"]
    
    def load_memory_variables(self, inputs: Dict) -> Dict[str, str]:
        # 将历史记录转换为字符串
        history_str = "\n".join([f"{item['role']}: {item['content']}" for item in self.history])
        return {"history": history_str}
    
    def save_context(self, inputs: Dict, outputs: Dict) -> None:
        # 保存对话历史
        self.history.append({"role": "user", "content": inputs["input"]})
        self.history.append({"role": "assistant", "content": outputs["response"]})
        
        # 如果超过最大长度，删除最早的记录
        while len(self.history) > self.max_length:
            self.history.pop(0)
    
    def clear(self) -> None:
        # 清空历史记录
        self.history = []

# 使用自定义Memory
memory = CustomMemory(max_length=5)
conversation = ConversationChain(llm=llm, memory=memory, verbose=True)

# 进行对话
conversation.predict(input="你好！")
conversation.predict(input="请介绍一下LangChain。")

---

六、核心概念：Agent

6.1 Agent的定义

Agent是LangChain中最高级别的抽象，它能够自主决策并执行任务。Agent通常包含以下组件：

1. LLM：核心大脑，用于推理和决策
2. Tools：工具集，用于与外部世界交互
3. Memory：记忆系统，用于存储对话历史
4. Prompt：提示模板，用于引导LLM

6.2 Agent类型

6.2.1 ZeroShotAgent

ZeroShotAgent不需要示例，直接根据工具描述进行决策。

from langchain.chat_models import ChatOpenAI
from langchain.agents import initialize_agent, AgentType
from langchain.tools import DuckDuckGoSearchRun, CalculatorTool

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

# 创建工具
tools = [
    DuckDuckGoSearchRun(name="搜索"),
    CalculatorTool(name="计算器")
]

# 创建Agent
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# 运行Agent
result = agent.run("搜索Python最新版本并计算发布至今的天数")
print(result)

6.2.2 ConversationalAgent

ConversationalAgent适用于对话场景，能够保持对话历史。

from langchain.chat_models import ChatOpenAI
from langchain.agents import initialize_agent, AgentType
from langchain.tools import DuckDuckGoSearchRun, CalculatorTool
from langchain.memory import ConversationBufferMemory

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

# 创建工具
tools = [
    DuckDuckGoSearchRun(name="搜索"),
    CalculatorTool(name="计算器")
]

# 创建Memory
memory = ConversationBufferMemory(memory_key="chat_history")

# 创建Agent
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION,
    memory=memory,
    verbose=True
)

# 进行多轮对话
result1 = agent.run("你好，我叫张三。")
print(result1)

result2 = agent.run("搜索今天北京的天气。")
print(result2)

result3 = agent.run("我明天要去北京出差，需要穿什么衣服？")
print(result3)

6.2.3 SelfAskWithSearchAgent

SelfAskWithSearchAgent会先进行自我提问，然后根据需要进行搜索。

from langchain.chat_models import ChatOpenAI
from langchain.agents import initialize_agent, AgentType
from langchain.tools import DuckDuckGoSearchRun

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

# 创建工具
tools = [DuckDuckGoSearchRun(name="搜索")]

# 创建Agent
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.SELF_ASK_WITH_SEARCH,
    verbose=True
)

# 运行Agent
result = agent.run("LangChain是什么时候发布的？")
print(result)

6.2.4 ReActAgent

ReActAgent结合了推理和行动，是最常用的Agent类型。

from langchain.chat_models import ChatOpenAI
from langchain.agents import initialize_agent, AgentType
from langchain.tools import DuckDuckGoSearchRun, CalculatorTool

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

# 创建工具
tools = [
    DuckDuckGoSearchRun(name="搜索"),
    CalculatorTool(name="计算器")
]

# 创建Agent
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.REACT_DOCSTORE,
    verbose=True
)

# 运行Agent
result = agent.run("计算圆周率的前10位")
print(result)

6.3 自定义Agent

from langchain.agents import AgentExecutor, BaseSingleActionAgent
from langchain.tools import DuckDuckGoSearchRun, CalculatorTool
from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from typing import List, Tuple, Any, Optional

class CustomAgent(BaseSingleActionAgent):
    llm: ChatOpenAI
    tools: List
    
    @property
    def input_keys(self) -> List[str]:
        return ["input"]
    
    def plan(
        self,
        intermediate_steps: List[Tuple[Any, str]],
        **kwargs: Any
    ) -> Any:
        # 获取输入
        input_text = kwargs["input"]
        
        # 使用LLM决定下一步行动
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"""你是一名Agent，拥有以下工具：
{[tool.name for tool in self.tools]}

请根据输入决定使用哪个工具或直接回答。

输出格式：
- 如果使用工具：TOOL_NAME: 参数
- 如果直接回答：直接输出答案"""),
            ("human", input_text)
        ])
        
        response = self.llm.predict(prompt.format())
        
        # 解析响应
        if ":" in response:
            tool_name, tool_input = response.split(":", 1)
            tool_name = tool_name.strip()
            tool_input = tool_input.strip()
            
            # 找到对应的工具
            for tool in self.tools:
                if tool.name == tool_name:
                    return tool, tool_input
        
        # 如果不使用工具，返回None
        return None, response
    
    async def aplan(
        self,
        intermediate_steps: List[Tuple[Any, str]],
        **kwargs: Any
    ) -> Any:
        return self.plan(intermediate_steps, **kwargs)

# 创建工具
tools = [
    DuckDuckGoSearchRun(name="搜索"),
    CalculatorTool(name="计算器")
]

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

# 创建自定义Agent
agent = CustomAgent(llm=llm, tools=tools)

# 创建Agent执行器
agent_executor = AgentExecutor.from_agent_and_tools(
    agent=agent,
    tools=tools,
    verbose=True
)

# 运行Agent
result = agent_executor.run("搜索今天的天气")
print(result)

---

七、核心概念：文档加载与处理

7.1 文档加载器

LangChain提供了多种文档加载器，可以从不同来源加载文档。

7.1.1 文本文件加载器

from langchain.document_loaders import TextLoader

# 创建加载器
loader = TextLoader("example.txt")

# 加载文档
documents = loader.load()

# 查看文档内容
for doc in documents:
    print(doc.page_content)

7.1.2 PDF文件加载器

# 安装依赖
pip install pypdf

from langchain.document_loaders import PyPDFLoader

# 创建加载器
loader = PyPDFLoader("example.pdf")

# 加载文档（按页加载）
documents = loader.load_and_split()

# 查看文档内容
for i, doc in enumerate(documents):
    print(f"第{i+1}页: {doc.page_content[:200]}")

7.1.3 Word文件加载器

# 安装依赖
pip install python-docx

from langchain.document_loaders import Docx2txtLoader

# 创建加载器
loader = Docx2txtLoader("example.docx")

# 加载文档
documents = loader.load()

# 查看文档内容
for doc in documents:
    print(doc.page_content)

7.1.4 网页加载器

from langchain.document_loaders import WebBaseLoader

# 创建加载器
loader = WebBaseLoader("https://langchain.com/")

# 加载文档
documents = loader.load()

# 查看文档内容
for doc in documents:
    print(doc.page_content[:500])

7.2 文档分割器

当文档过大时，需要将其分割成小块。

7.2.1 字符分割器

from langchain.text_splitter import CharacterTextSplitter

# 创建分割器
text_splitter = CharacterTextSplitter(
    separator="\n",
    chunk_size=1000,
    chunk_overlap=200
)

# 分割文本
text = """LangChain是一个用于构建基于LLM的应用程序的框架。
它提供了一套工具和抽象，使得开发者能够快速构建复杂的AI应用。
LangChain支持多种LLM提供商，包括OpenAI、Anthropic、Google等。
它还提供了丰富的工具集，包括搜索、计算、文件操作等。
LangChain的核心概念包括Chain、Prompt、Tool、Memory、Agent等。
通过这些概念，开发者可以构建各种类型的AI应用，如聊天机器人、问答系统、代码生成器等。"""

chunks = text_splitter.split_text(text)

# 查看分割结果
for i, chunk in enumerate(chunks):
    print(f"块{i+1}: {chunk}")

7.2.2 递归字符分割器

from langchain.text_splitter import RecursiveCharacterTextSplitter

# 创建分割器
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", " ", ""]
)

# 分割文本
chunks = text_splitter.split_text(text)

# 查看分割结果
for i, chunk in enumerate(chunks):
    print(f"块{i+1}: {chunk}")

7.2.3 文档分割器

from langchain.text_splitter import CharacterTextSplitter
from langchain.document_loaders import TextLoader

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

# 创建分割器
text_splitter = CharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)

# 分割文档
split_docs = text_splitter.split_documents(documents)

# 查看分割结果
for i, doc in enumerate(split_docs):
    print(f"文档块{i+1}: {doc.page_content[:200]}")

---

八、核心概念：向量数据库集成

8.1 向量数据库简介

向量数据库用于存储和检索嵌入向量，是构建RAG（Retrieval-Augmented Generation）应用的关键组件。

8.2 Chroma集成

# 安装依赖
pip install chromadb

from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter

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

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

# 创建向量数据库
embeddings = OpenAIEmbeddings()
vector_db = Chroma.from_documents(
    documents=split_docs,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# 持久化向量数据库
vector_db.persist()

# 加载向量数据库
vector_db = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings
)

# 进行相似性搜索
query = "LangChain的核心概念是什么？"
results = vector_db.similarity_search(query, k=3)

# 查看搜索结果
for i, result in enumerate(results):
    print(f"结果{i+1}: {result.page_content}")

8.3 Pinecone集成

# 安装依赖
pip install pinecone-client

from langchain.vectorstores import Pinecone
from langchain.embeddings import OpenAIEmbeddings
import pinecone

# 初始化Pinecone
pinecone.init(
    api_key="your-api-key",
    environment="us-west1-gcp"
)

# 创建索引（如果不存在）
index_name = "langchain-demo"
if index_name not in pinecone.list_indexes():
    pinecone.create_index(
        name=index_name,
        dimension=1536,
        metric="cosine"
    )

# 获取索引
index = pinecone.Index(index_name)

# 创建向量数据库
embeddings = OpenAIEmbeddings()
vector_db = Pinecone(
    index=index,
    embedding_function=embeddings.embed_query,
    text_key="text"
)

# 添加文档
from langchain.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter

loader = TextLoader("example.txt")
documents = loader.load()
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
split_docs = text_splitter.split_documents(documents)

vector_db.add_documents(split_docs)

# 进行相似性搜索
query = "LangChain的核心概念是什么？"
results = vector_db.similarity_search(query, k=3)

# 查看搜索结果
for i, result in enumerate(results):
    print(f"结果{i+1}: {result.page_content}")

8.4 FAISS集成

# 安装依赖
pip install faiss-cpu  # CPU版本
# pip install faiss-gpu  # GPU版本

from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter

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

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

# 创建向量数据库
embeddings = OpenAIEmbeddings()
vector_db = FAISS.from_documents(split_docs, embeddings)

# 保存向量数据库
vector_db.save_local("faiss_index")

# 加载向量数据库
vector_db = FAISS.load_local("faiss_index", embeddings)

# 进行相似性搜索
query = "LangChain的核心概念是什么？"
results = vector_db.similarity_search(query, k=3)

# 查看搜索结果
for i, result in enumerate(results):
    print(f"结果{i+1}: {result.page_content}")

---

九、核心概念：评估

9.1 评估概述

评估是衡量Agent性能的重要环节。LangChain提供了多种评估工具和方法。

9.2 使用LLM评估

from langchain.evaluation import load_evaluator
from langchain.chat_models import ChatOpenAI

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

# 创建评估器
evaluator = load_evaluator("qa", llm=llm)

# 定义问题和答案
question = "什么是LangChain？"
predicted_answer = "LangChain是一个用于构建AI应用的框架。"
reference_answer = "LangChain是一个用于构建基于LLM的应用程序的框架，它提供了一套工具和抽象，使得开发者能够快速构建复杂的AI应用。"

# 进行评估
result = evaluator.evaluate(
    prediction=predicted_answer,
    reference=reference_answer,
    input=question
)

# 查看评估结果
print(f"评估得分: {result['score']}")
print(f"评估理由: {result['reasoning']}")

9.3 使用标准评估数据集

from langchain.evaluation import load_evaluator
from langchain.chat_models import ChatOpenAI
from langchain.chains import LLMChain
from langchain.prompts import ChatPromptTemplate

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

# 创建Chain
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一名助手。"),
    ("human", "{question}")
])
chain = LLMChain(llm=llm, prompt=prompt)

# 创建评估器
evaluator = load_evaluator("qa", llm=llm)

# 定义测试数据集
test_cases = [
    {
        "question": "什么是Python？",
        "reference": "Python是一种高级、通用、解释型的编程语言，由Guido van Rossum于1991年创建。"
    },
    {
        "question": "什么是机器学习？",
        "reference": "机器学习是人工智能的一个分支，它使计算机能够从数据中学习并做出预测或决策。"
    },
    {
        "question": "什么是深度学习？",
        "reference": "深度学习是机器学习的一个子领域，它使用多层神经网络来学习数据的复杂特征。"
    }
]

# 进行评估
total_score = 0
for i, test_case in enumerate(test_cases):
    # 获取预测答案
    predicted_answer = chain.run(question=test_case["question"])
    
    # 进行评估
    result = evaluator.evaluate(
        prediction=predicted_answer,
        reference=test_case["reference"],
        input=test_case["question"]
    )
    
    total_score += result["score"]
    print(f"测试用例{i+1}:")
    print(f"  问题: {test_case['question']}")
    print(f"  预测答案: {predicted_answer}")
    print(f"  参考答案: {test_case['reference']}")
    print(f"  得分: {result['score']}")
    print(f"  理由: {result['reasoning']}")
    print()

# 计算平均得分
average_score = total_score / len(test_cases)
print(f"平均得分: {average_score}")

9.4 自定义评估器

from langchain.evaluation import EvaluatorType, load_evaluator
from langchain.evaluation.schema import Evaluator
from typing import Optional, Dict, Any

class CustomEvaluator(Evaluator):
    def evaluate(
        self,
        prediction: str,
        reference: Optional[str] = None,
        input: Optional[str] = None,
        **kwargs: Any
    ) -> Dict[str, Any]:
        # 自定义评估逻辑
        # 这里简单比较预测答案和参考答案的相似度
        
        if reference is None:
            return {"score": 0.5, "reasoning": "没有参考答案"}
        
        # 计算相似度（简单实现）
        prediction_words = set(prediction.lower().split())
        reference_words = set(reference.lower().split())
        
        if len(reference_words) == 0:
            return {"score": 0.0, "reasoning": "参考答案为空"}
        
        overlap = len(prediction_words & reference_words)
        similarity = overlap / len(reference_words)
        
        return {
            "score": similarity,
            "reasoning": f"预测答案与参考答案的相似度为{similarity:.2f}"
        }

# 使用自定义评估器
evaluator = CustomEvaluator()

# 进行评估
result = evaluator.evaluate(
    prediction="LangChain是一个用于构建AI应用的框架。",
    reference="LangChain是一个用于构建基于LLM的应用程序的框架。",
    input="什么是LangChain？"
)

print(f"评估得分: {result['score']}")
print(f"评估理由: {result['reasoning']}")

---

十、实战案例：构建知识库问答系统

10.1 需求分析

构建一个基于文档的知识库问答系统，用户可以上传文档并进行问答。

10.2 系统架构

┌─────────────────────────────────────────────────────────────┐
│                   知识库问答系统                              │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌─────────────┐    ┌─────────────┐    ┌─────────────┐     │
│  │  文档上传   │───▶│  文档处理   │───▶│  向量存储   │     │
│  │             │    │             │    │             │     │
│  └─────────────┘    └─────────────┘    └───────┬───────┘     │
│                                                │            │
│                                                ▼            │
│  ┌─────────────────────────────────────────────────────┐    │
│  │                    用户查询                          │    │
│  │           ┌───────────────────────────┐             │    │
│  │           │  问题解析 → 向量检索 → LLM生成  │             │    │
│  │           └───────────────────────────┘             │    │
│  └─────────────────────────────────────────────────────┘    │
│                                                             │
└─────────────────────────────────────────────────────────────┘

10.3 代码实现

from langchain.document_loaders import TextLoader, PyPDFLoader, Docx2txtLoader
from langchain.text_splitter import CharacterTextSplitter
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.chains import RetrievalQA
from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
import os
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

class KnowledgeBaseQA:
    """知识库问答系统"""
    
    def __init__(self, persist_directory="./chroma_db"):
        self.persist_directory = persist_directory
        self.embeddings = OpenAIEmbeddings()
        self.vector_db = None
        self.qa_chain = None
        self._initialize()
    
    def _initialize(self):
        """初始化向量数据库"""
        try:
            # 尝试加载现有的向量数据库
            self.vector_db = Chroma(
                persist_directory=self.persist_directory,
                embedding_function=self.embeddings
            )
            print("已加载现有的向量数据库")
        except:
            # 如果不存在，创建新的向量数据库
            self.vector_db = Chroma(
                persist_directory=self.persist_directory,
                embedding_function=self.embeddings
            )
            print("创建了新的向量数据库")
    
    def load_document(self, file_path):
        """加载文档"""
        # 根据文件类型选择加载器
        if file_path.endswith(".txt"):
            loader = TextLoader(file_path)
        elif file_path.endswith(".pdf"):
            loader = PyPDFLoader(file_path)
        elif file_path.endswith(".docx"):
            loader = Docx2txtLoader(file_path)
        else:
            raise ValueError(f"不支持的文件类型: {file_path}")
        
        # 加载文档
        documents = loader.load()
        print(f"已加载文档: {file_path}, 共{len(documents)}页")
        
        # 分割文档
        text_splitter = CharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200
        )
        split_docs = text_splitter.split_documents(documents)
        print(f"已分割为{len(split_docs)}个文档块")
        
        # 添加到向量数据库
        self.vector_db.add_documents(split_docs)
        self.vector_db.persist()
        print("文档已添加到向量数据库")
    
    def setup_qa_chain(self):
        """设置问答Chain"""
        llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
        
        # 创建自定义Prompt
        prompt = ChatPromptTemplate.from_messages([
            ("system", """你是一名专业的知识库问答助手。请根据提供的文档内容回答问题。

要求：
1. 仅基于提供的文档内容回答问题
2. 如果文档中没有相关信息，请明确说明
3. 回答要简洁明了，避免冗长
4. 对于复杂问题，提供详细的解释

文档内容：
{context}"""),
            ("human", "{question}")
        ])
        
        # 创建检索器
        retriever = self.vector_db.as_retriever(
            search_type="similarity",
            search_kwargs={"k": 3}
        )
        
        # 创建问答Chain
        self.qa_chain = RetrievalQA.from_chain_type(
            llm=llm,
            chain_type="stuff",
            retriever=retriever,
            chain_type_kwargs={"prompt": prompt},
            return_source_documents=True
        )
    
    def query(self, question):
        """查询知识库"""
        if self.qa_chain is None:
            self.setup_qa_chain()
        
        # 执行查询
        result = self.qa_chain({"query": question})
        
        # 返回结果
        return {
            "answer": result["result"],
            "sources": [doc.metadata.get("source", "") for doc in result["source_documents"]]
        }

# 使用示例
if __name__ == "__main__":
    # 创建知识库问答系统
    kb_qa = KnowledgeBaseQA()
    
    # 加载文档
    kb_qa.load_document("example.txt")
    kb_qa.load_document("example.pdf")
    
    # 查询知识库
    questions = [
        "LangChain的核心概念是什么？",
        "如何使用LangChain构建Agent？",
        "LangChain支持哪些LLM提供商？"
    ]
    
    for question in questions:
        print(f"\n问题: {question}")
        result = kb_qa.query(question)
        print(f"答案: {result['answer']}")
        print(f"来源: {result['sources']}")

10.4 运行结果分析

运行上述代码，系统会：

1. 加载指定的文档（支持txt、pdf、docx格式）
2. 将文档分割成小块并存储到向量数据库
3. 根据用户查询进行相似性检索
4. 将检索到的相关文档作为上下文输入给LLM
5. LLM基于上下文生成答案

---

十一、常见问题与解决方案

11.1 问题1：LLM调用失败

现象：调用LLM时出现错误。

原因分析：

• API密钥未正确配置
• 网络连接问题
• API调用频率超限

解决方案：

import os
from dotenv import load_dotenv
from langchain.chat_models import ChatOpenAI

# 加载环境变量
load_dotenv()

# 检查API密钥
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
    raise ValueError("OPENAI_API_KEY环境变量未设置")

# 创建LLM，增加重试次数
llm = ChatOpenAI(
    model="gpt-3.5-turbo",
    temperature=0,
    max_retries=3,
    request_timeout=60
)

# 测试调用
try:
    response = llm.predict("Hello")
    print(response)
except Exception as e:
    print(f"调用失败: {e}")

11.2 问题2：文档分割不合理

现象：文档分割后，相关内容被分割到不同的块中。

原因分析：

• 分割器配置不当
• 分割粒度不合适

解决方案：

from langchain.text_splitter import RecursiveCharacterTextSplitter

# 使用递归分割器，按优先级选择分隔符
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=[
        "\n\n",  # 优先按段落分割
        "\n",    # 其次按行分割
        "。",    # 中文句号
        "！",    # 中文感叹号
        "？",    # 中文问号
        ".",     # 英文句号
        " ",     # 空格
        ""       # 最后按字符分割
    ]
)

# 分割文档
split_docs = text_splitter.split_documents(documents)

11.3 问题3：向量检索结果不准确

现象：检索到的文档与查询不相关。

原因分析：

• 嵌入模型选择不当
• 检索参数配置不合理
• 文档质量差

解决方案：

from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings

# 使用合适的嵌入模型
embeddings = OpenAIEmbeddings(model="text-embedding-ada-002")

# 创建向量数据库
vector_db = Chroma.from_documents(
    documents=split_docs,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# 调整检索参数
retriever = vector_db.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={
        "k": 5,
        "score_threshold": 0.7
    }
)

# 进行检索
results = retriever.get_relevant_documents("查询内容")

11.4 问题4：Agent工具调用失败

现象：Agent无法正确调用工具。

原因分析：

• 工具描述不清晰
• Prompt模板不合适
• LLM无法理解工具用途

解决方案：

from langchain.tools import BaseTool
from typing import Optional, Type
from pydantic import BaseModel, Field

class SearchInput(BaseModel):
    query: str = Field(description="搜索查询，用于查找相关信息")

class CustomSearchTool(BaseTool):
    name = "web_search"
    description = """用于搜索互联网上的信息。
    当用户的问题需要最新的信息、事实性知识或外部数据时，应该使用此工具。
    例如：查询天气、搜索新闻、查找技术文档等。
    输入参数：query - 搜索关键词"""
    
    args_schema: Type[BaseModel] = SearchInput
    
    def _run(self, query: str) -> str:
        # 搜索逻辑
        return f"搜索结果: {query}"

11.5 问题5：Memory占用过大

现象：随着对话进行，Memory占用越来越大，导致性能下降。

原因分析：

• 对话历史不断累积
• 没有设置存储上限

解决方案：

from langchain.memory import ConversationBufferWindowMemory
from langchain.memory import ConversationSummaryMemory
from langchain.chat_models import ChatOpenAI

# 方法1：使用窗口Memory，限制历史记录数量
memory = ConversationBufferWindowMemory(k=10)

# 方法2：使用摘要Memory，将历史记录压缩为摘要
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
memory = ConversationSummaryMemory(llm=llm)

# 方法3：定期清理Memory
memory.clear()

---

十二、总结

本章详细介绍了LangChain的核心概念，包括：

1. Prompt：引导LLM生成特定输出的文本模板
2. Chain：一系列组件的组合，用于处理输入并生成输出
3. Tool：Agent与外部世界交互的桥梁
4. Memory：存储对话历史，实现上下文理解
5. Agent：最高级别的抽象，能够自主决策并执行任务
6. 文档加载与处理：从不同来源加载和处理文档
7. 向量数据库集成：存储和检索嵌入向量，构建RAG应用
8. 评估：衡量Agent性能的方法和工具

通过实战案例，我们展示了如何使用LangChain构建一个完整的知识库问答系统，涵盖了文档加载、向量存储、检索和生成等核心功能。

LangChain的模块化设计和丰富的集成能力使其成为构建Agent应用的首选框架。掌握LangChain的核心概念，将为后续的Agent开发打下坚实的基础。

---

练习题

1. 基础题：使用LangChain创建一个简单的对话Chain，能够回答用户问题。
2. 进阶题：使用LangChain创建一个多步骤的SequentialChain，完成文章生成任务。
3. 实战题：创建一个工具调用Agent，能够使用搜索和计算器工具回答问题。
4. 挑战题：构建一个完整的RAG系统，支持文档上传、向量存储和问答功能。