3.2 外协对接：与外部工具库集成

3.2 外协对接：与外部工具库集成

本节摘要：单个工具自制够了，但搜索、数据库、云服务这类成建制的外部能力要走集成路线。本节讲四类外协对接的标准姿势——搜索与计算工具进代理、Redis 做分布式记忆、托管向量库做共享仓库，并给出集成前的检查清单。

外协的三种典型形态

装配线不可能什么都自己造。外部能力的接入按"对方提供什么"分三类：提供 API 的（搜索、行情、翻译），包成工具进代理工位；提供存储的（Redis、托管向量库），替换记忆与仓库的底层实现；提供完整功能的库（计算、解析），直接做工具内核。三类对应三套对接姿势，先看最常用的搜索加计算组合：

from langchain.agents import initialize_agent, AgentType, load_tools
from langchain_openai import OpenAI
import os

llm = OpenAI(temperature=0)

# 搜索与计算是两个经典外协件
# 搜索需要环境变量里有搜索服务的密钥
tools = load_tools(["serpapi", "llm-math"], llm=llm)

agent = initialize_agent(
    tools, llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True,
)
print(agent.run("北京今天最高气温多少摄氏度？乘以2再加10等于多少"))
# verbose 输出（节选）：
# Thought: 需要先查今天的气温
# Action: Search
# Action Input: 北京今天最高气温
# Observation: 31℃
# Thought: 再做计算
# Action: calculator
# Action Input: 31*2+10
# Observation: 72
# Final Answer: 今天最高约31度，计算结果是72
​

姿势一：把任意 API 包成外协件

搜索有现成适配，业务里的内部接口就得自己包。用 requests 库对接一个内部订单系统：

import requests
from langchain_core.tools import tool

@tool
def query_order(order_id: str) -> str:
    """根据订单号查询物流状态。用户询问订单进度或物流信息时使用。"""
    try:
        # 对接内部订单接口（演示用公开测试接口）
        resp = requests.get(
            "https://httpbin.org/get",
            params={"order": order_id}, timeout=5)
        data = resp.json()
        # 实际项目在这里解析业务字段
        return f"订单 {order_id} 查询成功，接口已响应"
    except requests.Timeout:
        return "订单接口超时，请稍后再试"

print(query_order.invoke({"order_id": "SO-1024"}))
# 输出：订单 SO-1024 查询成功，接口已响应
​

外协件的四条军规全部体现在这段代码里：超时必须设（外协不可控，别让它卡死产线）；异常要捕获并转成可读文本（代理需要"知道失败"而非"直接停线"）；返回值简短（它会进上下文）；描述写清使用时机（3.1 节军规依然有效）。

姿势二：记忆搬到分布式仓库

2.5 节的记忆在进程内存里，重启即失、多副本不共享。外协给 Redis 一份就解决：

from langchain.memory import ConversationBufferMemory
from langchain_community.memory import RedisChatMessageHistory
from langchain.chains import ConversationChain
from langchain_openai import OpenAI

# 分布式记录本：历史存进 Redis，多副本共享
history = RedisChatMessageHistory(
    session_id="user-1024", url="redis://localhost:6379")

memory = ConversationBufferMemory(
    chat_memory=history,
    memory_key="history",
    return_messages=False,
)

conversation = ConversationChain(
    llm=OpenAI(temperature=0), memory=memory, verbose=False)

print(conversation.predict(input="我叫小舟，做彩色袜子生意"))
print(conversation.predict(input="我做什么生意"))
# 第二问输出示例：你经营彩色袜子生意。
# 换一个进程再开一个同 session 的会话，历史依然在
​

替换动作只发生在构造参数上，链的组装一字不改——这就是 2.5 节说的"记录本可替换"兑现的时刻。

姿势三：仓库搬到托管服务

向量库同理。本地 Chroma 换成云托管，代码改动集中在一处：

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

texts = ["支持七天无理由退货", "棉袜建议冷水手洗"]

# 本地仓库
local_vs = Chroma.from_texts(texts, OpenAIEmbeddings())
print(local_vs.similarity_search("退货", k=1)[0].page_content)
# 输出：支持七天无理由退货

# 换托管服务时：只换这一行的构造，其余代码不动
# from langchain_community.vectorstores import Pinecone
# cloud_vs = Pinecone.from_texts(
#     texts, OpenAIEmbeddings(), index_name="socks-kb")
​

四类集成对象的注意点集中一张表：

⚠️ 外协密钥管理的两个高频事故：密钥写进代码进了版本库；测试密钥带到生产环境忘了换。密钥一律环境变量注入，启动时打印脱敏确认。

💡 集成的成本核算要看"故障半径"：多一个外协，产线就多一个可能宕机的上游。给每个外协件想清楚降级路径——搜索挂了能否退回知识库作答、订单接口超时能否回复"稍后查询"，这比接通它更重要。

集成前检查清单

任何外部能力接进产线前过五问：有没有官方适配包（有就不自己包）；密钥怎么注入与轮换；超时与重试策略是什么；失败时代理看到什么信息；这个外协挂掉时产线怎么降级。五问答完，这个外协件才真正"可上线"。

下一节进入本章枢纽：产线改完了，怎么用数据证明它变好了——评估与测试。