问答

怎么给 LangChain / LangGraph 智能体加上社媒数据工具?

简短回答

两种做法:把需要的接口包成 LangChain 工具,通过 HTTP 调 okdata;或者用 langchain-mcp-adapters 加载 okdata 的 MCP 服务,一次拿到覆盖全部 51 个接口的 find、describe、call 工具。只有取数据的调用计费,比如小红书搜索每页 $0.02,X 搜索每页 $0.006。

选哪种,看任务有多开放。固定流程,比如每天早上在小红书搜三个产品名、总结新笔记,最适合手写一两个工具:提示词短,调用可预期,模型没什么可探索的。开放式的调研智能体,今天要抖音、明天要 X,用 MCP 服务更合适,它可以自己发现有哪些接口、各要多少钱。

做法一:手写工具

import os, requests
from langchain_core.tools import tool

API = "https://api.okdata.online"
H = {"Authorization": f"Bearer {os.environ['OKDATA_API_KEY']}"}

@tool
def search_xiaohongshu(query: str, sort: str = "general") -> list[dict]:
    """搜索小红书公开笔记。sort 可选 general、latest、popular。
    每次调用返回约 20 篇,计费一次。"""
    body = requests.get(API + "/v1/xiaohongshu/notes/search", headers=H, timeout=60,
                        params={"query": query, "sort": sort,
                                "fields": "id,url,title,text,metrics.likes,author.name"}).json()
    if body["code"] != "ok":
        # 把错误交给模型,让它改参数或者停下
        return [{"error": body["code"], "message": body["message"], "retry": body["retry"]}]
    return body["data"]["items"]

出错时把错误当结果返回,别直接抛异常。retry 字段会告诉模型值不值得重试,而且失败的调用不扣费。背后的接口是小红书笔记搜索;抖音视频搜索和 X 帖子搜索写法一样,换个路径即可。

做法二:用 langchain-mcp-adapters 加载 MCP 服务

import os
from langchain_mcp_adapters.client import MultiServerMCPClient

client = MultiServerMCPClient({
    "okdata": {
        "transport": "streamable_http",
        "url": "https://api.okdata.online/mcp",
        "headers": {"Authorization": f"Bearer {os.environ['OKDATA_API_KEY']}"},
    }
})
tools = await client.get_tools()  # find、describe、call、account

把 tools 传给 create_agent 或 LangGraph 的 ToolNode,和普通工具一样用。find、describe 不收费,智能体可以先看清价格再调用。

一轮调研花多少

调研内容 调用次数 费用
小红书搜 3 次 3 $0.06
抖音搜 3 次 3 $0.06
X 搜索翻 5 页 5 $0.03
以上混在一轮里 11 {{cost:xiaohongshu.notes.search3+douyin.videos.search3+x.posts.search*5}}

让它无人值守之前要设好的

  • 步数上限。 LangGraph 的 recursion_limit(旧版智能体是 max_iterations)是防止一路翻页最便宜的保险。
  • 结果要小。 调用时带 fields,每次工具结果控制在几 KB;见 view 和 fields 的说明。
  • 文本不可信。 文案、笔记、评论都是陌生人写的。告诉模型这些只是数据;会发消息、会花钱的工具,前面加一道人工确认。
  • 重试要安全。 走 HTTP 时带 Idempotency-Key,走 MCP 时给 call 传 idempotency_key。24 小时内重复请求直接返回已存结果,不再扣费。

其他选择

LangChain 社区自带的加载器,比如 YouTube 那个,免费,在自己电脑上跑没问题,但不少底层依赖抓取库,到云服务器上容易失效或被封 IP。平台有官方接口而且够用时,官方接口是首选,比如用 YouTube Data API 取搜索元数据。抖音、小红书没有面向公众的官方数据接口可以退回去用。其他框架的接法见 MCP 页。

更新于