LangChain 开发自定义组件_系列12

转载：小红书 AI产品赵哥

前言🔖

到今天，我们已经一起走过了很长的路。我们从零开始，学会了用 LangChain 构建应用，掌握了高级 RAG 架构，攻克了性能优化和工程排坑的难题，甚至还让 AI 学会使用工具，学会了理解多模态世界。可以说，你们现在看到的 LangChain，是一个拥有海量武器的巨大仓库。

但任何一个资深的指挥官都知道，为自己量身定做的才是最趁手的武器。预制的武器再好，也无法完全贴合你所在企业的个性化场景：

你的用户数据，可能躺在一个没人敢动的 Oracle 老数据库里。
你的实时库存信息，需要通过一个内部的 RPC 接口才能查询。
你的项目管理流程，深度绑定在一个自研的 Jira 插件上。

LangChain 提供的几百个包里，没有这些。那么，游戏结束了吗？

恰恰相反，这才是游戏真正的开始。LangChain 最强大的特性，不是它提供了什么，而是它允许你创造什么。今天，我们来聊聊：自定义扩展开发。我们将亲自为 AI 打造全新的 “手臂” 和 “记忆”，让它能够无缝地接入你公司个性化的系统里。

准备好了吗？咱们发车了，走起。。。

一、为什么标准组件不够用？🔖

在动手之前，我们必须先统一思想：为什么要费力去自定义？用现成的不香吗？

LangChain 的集成生态，解决的是通用问题。比如，连接 MySQL，连接 Salesforce，连接 Google Drive。这些都是标准化的、面向公众的服务。

但企业的核心竞争力，往往体现在那些非标的、私有的系统中。这些系统沉淀了公司多年的业务逻辑和数据资产。能够将 AI 能力与这些系统深度融合，才能创造出真正的护城河。自定义让你能做到：

让 AI 客服能查询一个内部订单管理系统，实时回答用户的物流状态。
让 AI 销售助手能操作公司的 CRM 系统，自动为新线索创建跟进任务。
让 AI 数据分析师能连接一个私有的数据仓库，执行特定的 SQL 查询并生成报告。

在 LangChain 中，实现这种连接，最主要通过自定义两种组件：

自定义工具 (Tool)：为你的 Agent 创造新的行动能力。让它能做一些事，比如调用一个 API、修改一个数据库条目。
自定义检索器 (Retriever)：为你的 RAG 系统提供新的知识来源。让它能从任何地方获取信息，无论那个地方是 SQL 数据库、一个内部 wiki，还是一个 Elasticsearch 集群。

二、自定义 Tool：让 AI 大脑有手有脚！🔖

我们已经知道，Agent 的威力取决于它拥有多少工具。但工具到底是什么？从 Agent 的视角看，一个工具包含三要素：

名称 (name)：一个独一无二的标识符，比如 query_order_status。
描述 (description)：这是给 LLM 看的，是所有环节中最最最重要的一部分！它告诉 LLM，这个工具是干什么的，什么时候应该用它，它需要什么参数。
参数结构 (args_schema)：定义了这个工具需要哪些输入，以及每个输入的类型。

LangChain 提供了两种创建自定义工具的方式：一种简单快捷，一种灵活强大。

🔹2.1. 简单方式：使用 @tool 装饰器

这种方式比较简单，推荐入门使用。你只需要写一个普通的 Python 函数，然后用 @tool 装饰一下，LangChain 就会自动帮你搞定上面说的三要素。

名称：默认就是你的函数名。
描述：自动取自你的函数的文档字符串（docstring）。
参数结构：自动根据你的函数的类型提示（type hints）来生成。

实战场景：创建一个查询内部员工信息的工具

假设你的公司有一个内部 API，可以通过员工姓名查询其邮箱和所属部门。我们需要为 HR 智能助手 Agent 创建一个能调用这个 API 的工具。

第一步：编写核心逻辑函数（模拟 API 调用）

# 这是一个模拟的内部员工数据库
_INTERNAL_EMPLOYEE_DB = {
    "张三": {"email": "zhang.san@mycompany.com", "department": "技术部"},
    "李四": {"email": "li.si@mycompany.com", "department": "产品部"},
}

def _query_employee_api(employee_name: str) -> str:
    """这是一个模拟的API调用函数"""
    if employee_name in _INTERNAL_EMPLOYEE_DB:
        return str(_INTERNAL_EMPLOYEE_DB[employee_name])
    else:
        return f"找不到名为'{employee_name}'的员工。"

第二步：使用 @tool 封装成工具

from langchain.tools import tool

@tool
def query_employee_info(employee_name: str) -> str:
    """
    查询内部员工的邮箱和所属部门。
    参数:
        employee_name: 员工姓名 (str)
    返回:
        员工信息字符串或未找到提示
    """
    return _query_employee_api(employee_name)



@tool
def employee_info_search(employee_name: str) -> str:
    """
    当需要查询公司内部员工的邮箱或所属部门时，使用此工具。
    你需要提供员工的完整姓名作为输入。
    例如，输入'李四'，就可以找到他的联系信息。
    """
    # 在工具内部，调用我们刚才写的核心逻辑
    return _query_employee_api(employee_name)

💡 产品同学请注意：

上面那个长长的文档字符串，不是写给程序员看的，是写给 LLM 看的。你必须用自然语言清晰地告诉它：这个工具是干嘛的？什么时候用？需要我给你什么？ 你的描述越清晰，Agent 就越能领会你的意思，在恰当的时候调用它。这个描述，就是你为这个工具功能所撰写的产品需求规格说明书。

第三步：在 Agent 中使用这个新工具

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder

# 初始化模型
llm = ChatOpenAI(model="gpt-4o")

# 定义工具列表
tools = [employee_info_search]

# 构建提示词模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个HR智能助手，负责查询员工信息。"),
    ("user", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

# 创建Agent
agent = create_openai_tools_agent(llm, tools, prompt)

# 执行Agent
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

当你运行这段代码，并打开 verbose=True 时，你就能看到 Agent 的思考过程：

[Agent] 思考：用户想找 “李四的邮箱”。我检查了一下我的工具箱，发现一个叫 employee_info_search 的工具，它的描述说可以查询员工的邮箱和部门，需要输入员工姓名。
- 以 “查询公司内部员工的邮箱”。太棒了，正好能用。这个工具需要一个 employee_name 参数，我从用户的问题里提取出来了，就是 “李四”。
[Agent] 调用：Invoking: employee_info_search with {'employee_name': '李四'}
[Tool]（employee_info_search 函数被执行，返回了结果字符串）
[Agent] 思考：我拿到了工具的返回结果：{'email': 'li.si@mycompany.com', 'department': '产品部'}。现在我要把这个结果用友好的方式告诉用户。
[Agent] 生成最终回答：“当然可以，李四的邮箱是 li.si@mycompany.com，他属于产品部。”

🔹2.2. 高级方式：继承 BaseTool 类

@tool 装饰器虽然方便，但有其局限性。当你需要更精细的控制时，比如：

需要分别定义同步执行（_run）和异步执行（_arun）的逻辑。
工具的初始化过程比较复杂，需要在创建时传入一些配置。
需要更严格地定义输入参数的结构和验证规则。

这时，你就需要通过继承 BaseTool 来创建一个类。

实战场景：创建一个查询实时产品库存的工具

第一步：使用 Pydantic 定义严格的输入参数

from pydantic import BaseModel, Field

class ProductInventoryInput(BaseModel):
    product_sku: str = Field(description="必须提供产品的标准库存单位（SKU），例如 'TSHIRT-RED-L'。")

使用 Pydantic 的好处是，它不仅能自动生成参数的 JSON Schema 给 LLM 看，还能在工具被调用时，自动验证传入的参数是否合法。

第二步：创建工具类

from typing import Type
from langchain.tools import BaseTool

def _query_inventory_db(sku: str) -> int:
    """模拟查询数据库，返回库存数量"""
    # 实际场景这里是连接数据库的复杂逻辑
    if sku == "TSHIRT-RED-L":
        return 120
    else:
        return 0


class ProductInventoryTool(BaseTool):
    name: str = "realtime_product_inventory_check"
    description: str = "用于查询特定产品SKU的实时库存数量。这个工具非常精准，直接连接生产数据库。"
    args_schema: Type[BaseModel] = ProductInventoryInput

    def _run(self, product_sku: str) -> str:
        """工具的同步执行逻辑"""
        inventory_count = _query_inventory_db(product_sku)
        return f"产品SKU '{product_sku}' 的当前库存为: {inventory_count} 件。"

    async def _arun(self, product_sku: str) -> str:
        """工具的异步执行逻辑（可选）"""
        # 在真实应用中，这里可能是异步的数据库客户端调用
        return self._run(product_sku)  # 为简化，我们直接调用同步逻辑

第三步：在 Agent 中使用

使用方法和之前完全一样，只是在创建 tools 列表时，需要实例化这个类：

tools = [ProductInventoryTool()]

💡 产品经理的核心工作：

对于自定义工具，你的核心产出物，就是一份清晰的工具规格说明书。它需要明确：

工具的业务目标：这个工具要解决什么业务问题？
工具的名称：一个简洁、表意的英文名。
给 LLM 看的描述：这是重中之重，参考我们上面的例子。
输入参数定义：需要哪些参数？每个参数的含义是什么？是否必须？是什么类型？
输出结果定义：成功时返回什么？失败时返回什么？格式是怎样的？

这份文档，就是连接你的产品构想和技术实现的桥梁。

三、自定义 Retriever：不仅是查向量库

我们知道，RAG 的核心是检索器（Retriever）。LangChain 内置了各种向量数据库的检索器。但如果你的知识源根本不是一个向量数据库呢？

比如，你的知识就存在一个 SQL 数据库的特定表里，或者需要通过一个内部的 wiki API 来搜索。这时，我们就需要自定义一个检索器。

🔹自定义检索器的本质

一个检索器，无论多复杂，其核心就是一个函数：_get_relevant_documents。这个函数接收一个字符串查询（用户的提问），返回一个 Document 对象的列表。

💡 所以，自定义一个检索器，本质上就是写一段你自己的逻辑，去你指定的任何地方，根据一个查询字符串，拿到一些相关的文本，然后把这些文本包装成 LangChain 认识的 Document 格式。

🔹实战场景：创建一个从 SQL 数据库检索公司新闻的检索器

假设公司所有的新闻稿都存在一个 SQL 数据库的 news 表中，包含 id，title，content，publish_date 等字段。我们希望 AI 能直接在这个表里搜索新闻。

第一步：准备数据源（模拟 SQL 数据库）

import sqlite3

# 创建一个内存中的SQLite数据库用于演示
conn = sqlite3.connect(":memory:", check_same_thread=False)
cursor = conn.cursor()

# 创建表
cursor.execute("""
CREATE TABLE news (
    id INTEGER PRIMARY KEY,
    title TEXT NOT NULL,
    content TEXT NOT NULL,
    publish_date TEXT NOT NULL
)
""")
conn.commit()

ontent TEXT NOT NULL,
publish_date TEXT NOT NULL
)
""")

# 插入一些假数据
news_data = [
    (1, "公司发布AI战略", "在今天的发布会上，CEO张三宣布了公司全面的AI战略...", "2024-05-10"),
    (2, "新产品'凤凰'发布", "备受期待的新产品'凤凰'今日正式发布，它集成了业界领先的AI能力...", "2024-05-20"),
    (3, "季度财报表现强劲", "公司公布了第一季度财报，营收同比增长20%...", "2024-04-28"),
]
cursor.executemany("INSERT INTO news VALUES (?, ?, ?, ?)", news_data)
conn.commit()

第二步：创建自定义检索器类

import sqlite3
from typing import List
from langchain_core.retrievers import BaseRetriever
from langchain_core.callbacks import CallbackManagerForRetrieverRun
from langchain_core.documents import Document


class SqlNewsRetriever(BaseRetriever):
    """
    自定义SQL新闻检索器
    从SQLite数据库的news表中检索与查询相关的新闻内容，并封装为LangChain的Document对象
    """
    conn: sqlite3.Connection  # 接收一个数据库连接作为参数

    def _get_relevant_documents(
        self, query: str, *, run_manager: CallbackManagerForRetrieverRun
    ) -> List[Document]:
        """
        自定义的核心检索逻辑
        :param query: 用户的查询字符串
        :param run_manager: 回调管理器（LangChain内置参数）
        :return: 封装后的Document对象列表
        """
        print(f"--- 正在用SQL检索与 '{query}' 相关的新闻 ---")
        cursor = self.conn.cursor()

        # 简单的模糊匹配检索逻辑（实际场景可替换为更复杂的检索策略）
        # 比如结合LLM将自然语言query转为精准SQL，或接入全文搜索引擎
        sql_query = "SELECT * FROM news WHERE title LIKE ? OR content LIKE ?"
        like_query = f"%{query}%"

        cursor.execute(sql_query, (like_query, like_query))
        rows = cursor.fetchall()

        # 将SQL查询结果包装成LangChain的Document对象列表
        documents = []
        for row in rows:
            # row 结构: (id, title, content, publish_date)
            doc = Document(
                page_content=f"标题: {row[1]}\n内容: {row[2]}",
                metadata={
                    "source": "Internal News DB",  # 标记数据来源
                    "news_id": row[0],             # 新闻ID
                    "publish_date": row[3]         # 发布日期
                }
            )
            documents.append(doc)

        return documents

💡 产品同学请注意：_get_relevant_documents 这个方法里的逻辑，就是你和数据工程师需要一起设计的 “黑盒子”。而 Document 的创建过程，尤其是 metadata 的填充，是你需要关注的。丰富的元数据，能为后续的 RAG 流程提供宝贵的上下文信息。

第三步：在 RAG 链中使用自定义检索器

from langchain.chains import RetrievalQA

# 1. 实例化我们的自定义检索器
sql_retriever = SqlNewsRetriever(conn=conn)

# 2. 像使用任何其他检索器一样，把它用在RAG链里
qa_chain = RetrievalQA.from_chain_type(
    llm=ChatOpenAI(model="gpt-4o"),
    chain_type="stuff",
    retriever=sql_retriever
)

# 3. 开始提问!
response = qa_chain.invoke("关于公司AI战略的新闻讲了些什么?")
print(response["result"])

# 内部流程:
# 1. "公司AI战略" 这个查询被传入 SqlNewsRetriever。
# 2. `_get_relevant_documents` 被触发，执行SQL查询，找到ID为1的新闻。
# 3. 该新闻被包装成一个Document对象。
# 4. 这个Document对象和问题一起被送给LLM。
# 5. LLM根据新闻内容，总结并回答了问题。

四、进阶：让自定义组件协同工作

现在，我们拥有了自定义的工具和自定义的检索器。真正的威力在于将它们组合起来，放入我们之前讨论过的高级 Agent 或路由架构中。

想象一个 CEO 的 AI 作战室。CEO 可以提出一个非常开放的问题：

我们的新产品发布后，市场反应怎么样？
另外，帮我看看销售团队的负责人李四的联系方式，我要马上联系他了解一线情况。

一个集成了我们自定义组件的终极 Agent 会这样工作：

[Agent 思考]：这个问题包含两个部分。第一部分是市场反应，这需要查询知识。第二部分是找 “李四” 的联系方式，这需要一个行动工具。
[Agent 调用检索工具]：Agent 知道它有一个可以检索新闻的工具（这个工具就是把我们的 SqlNewsRetriever 包装了一下）。它调用这个工具，搜索 “市场反应”。
[自定义检索器工作]：SqlNewsRetriever 执行 SQL 查询，找到了发布的新闻稿，并作为上下文返回。
[Agent 调用行动工具]：Agent 接着处理第二部分。它看到 “李四的联系方式”，想起了它的工具箱里有 employee_info_search 这个神器。
[自定义工具工作]：employee_info_search 工具被调用，查询内部 API，返回了李四的邮箱和部门。
[Agent 合成答案]：Agent 现在手握两份情报：新闻稿内容，和李四的联系方式。它将这两部分信息整合在一起，生成最终的、全面的报告，呈现给 CEO。

这就是 LangChain 可组合性（Composability） 的终极体现。每一个你自定义的组件，都像一个乐高积木，可以无缝地嵌入到任何复杂的架构中，共同协作，完成仅靠通用组件无法想象的复杂任务。

五、来，总结一下吧

今天，我们深入了解了 LangChain，学会了如何为它加装定制的零件。我希望你们能深刻理解，自定义开发，不是一个纯粹的技术活，它是一个深刻的产品创新过程。

自定义工具，是在为你的 AI 定义新的 “动词”。它能 “做什么”？是查询、是创建、是更新，还是通知？
自定义检索器，是在为你的 AI 连接新的 “大脑海马体”。它能 “知道什么”？是结构化的数据库，是非结构化的文档，还是实时的事件流？

作为产品经理，你的新角色，就是成为公司内部各种系统和数据源的 **“AI 封装架构师”**。你需要：

盘点资产：梳理出公司内部最有价值的、但尚未被 AI 触及的数据孤岛和系统能力。
定义接口：为这些资产设计 “AI 友好” 的工具或检索器接口，也就是我们反复强调的 “规格说明书”。
规划整合：思考如何将这些新的自定义组件，编排进你的 AI 产品架构中，去解决真实、具体的业务痛点。

当你能够引导你的团队，为 AI 装上这些独一无二的 “手脚” 和 “记忆” 时，你所打造的，将不再是一个千篇一律的 GPT 套壳应用，而是一个深度嵌入企业血脉、拥有强大护城河、真正不可替代的智能解决方案。

92IT

Just love IT.

LangChain 开发自定义组件_系列12