1. 为什么要在 crewAI 里复用 LangChain 工具
如果你已经用 LangChain 攒了一堆工具——数据库查询、API 封装、文档处理、搜索抓取——现在想切到 crewAI 做多智能体编排,第一反应往往是"难道要全部重写一遍"。我一开始也这么担心,实际跑下来发现完全不用。crewAI 提供了LangChainTool适配器,能把 LangChain 的BaseTool实例直接包成 crewAI 能识别的工具对象,Agent 拿过去就能用。
这件事的价值在于分工。crewAI 的强项是角色定义、任务调度、多 Agent 协作流程;LangChain 的强项是工具生态和 Chain 抽象,社区里现成的工具数量非常可观。硬要 crewAI 自己重造所有工具,既费时间又容易踩坑。让 crewAI 管编排、LangChain 管工具,两边各干各擅长的事,链路反而更稳。
这篇聚焦一个具体问题:怎么把 LangChainTool 适配进 crewAI Agent,并且在链式复用场景里保持参数和返回值一致。所谓链式复用,指的是一个 LangChain Chain(比如 LCEL 表达式或 RetrievalQA)被封装成 crewAI 工具后,Agent 调用它、拿到结构化结果、再传给下一个 Agent 或下一个工具,中间不能出现参数名对不上、返回值类型漂移的情况。我会给出可复制的适配代码、依赖版本、最小验证步骤,以及几个真实报错的排查方法。适合已经写过 LangChain 工具、想迁移到 crewAI 多智能体架构的开发者。
在开始之前,先明确一个前置条件:你需要一个能稳定调用模型的 API 入口。crewAI 和 LangChain 本身只是编排框架,真正干活的是背后的 LLM。我这边习惯用 TaoToken 做统一接入,它的 Base URL 兼容 OpenAI 协议,LangChain 的ChatOpenAI和 crewAI 的 LLM 配置都能直接指过去,省得每个框架单独配一套密钥。下面第二节会讲具体怎么接。
2. TaoToken 前置:统一模型入口与依赖版本
跨框架互操作最容易出问题的地方不是适配器本身,而是两个框架各自去连模型时配置不一致。LangChain 用ChatOpenAI,crewAI 用自己的 LLM 封装,如果两边指向不同的 endpoint 或不同的模型 ID,调试时你会分不清是适配器的问题还是模型的问题。所以第一步先把模型入口统一。
TaoToken 的 API 地址是https://taotoken.net/api,兼容 OpenAI 的/v1/chat/completions协议。你需要在控制台创建一个 API Key,然后把它设成环境变量。我建议用OPENAI_API_KEY和OPENAI_API_BASE这两个标准变量名,因为 LangChain 和 crewAI 默认都会读它们,这样两边不用各写一套配置。
export OPENAI_API_KEY="sk-你的taotoken密钥" export OPENAI_API_BASE="https://taotoken.net/api"依赖版本这块要卡死,跨框架适配对版本很敏感。我实测下来这套组合能跑通:
pip install "crewai==1.11.0" \ "langchain==0.3.7" \ "langchain-core==0.3.15" \ "langchain-community==0.3.5" \ "langchain-openai==0.2.6" \ "pydantic==2.9.2"crewAI 1.11.0 的LangChainTool在crewai.tools下,LangChain 0.3.x 的BaseTool接口和 0.2.x 有差异,混用会出现args_schema校验失败。如果你之前装过旧版,先pip uninstall干净再装。验证版本:
import crewai, langchain, langchain_core print(crewai.__version__) # 1.11.0 print(langchain.__version__) # 0.3.7 print(langchain_core.__version__) # 0.3.15模型 ID 方面,TaoToken 支持主流模型,我在示例里用gpt-4o-mini做默认,因为它在工具调用场景下响应快、成本低。你可以在模型对话页面先确认目标模型 ID 拼写,避免因为模型名写错导致 404。LangChain 侧配置:
from langchain_openai import ChatOpenAI llm = ChatOpenAI( model="gpt-4o-mini", base_url="https://taotoken.net/api", api_key="sk-你的taotoken密钥", temperature=0, )crewAI 侧配置:
from crewai import LLM llm = LLM( model="openai/gpt-4o-mini", base_url="https://taotoken.net/api", api_key="sk-你的taotoken密钥", )注意 crewAI 的model字段要带openai/前缀,这是它区分 provider 的方式;LangChain 的ChatOpenAI不需要前缀。这个差异是新手最容易踩的坑之一,两边配置看起来像但格式不同。把这两个 LLM 对象分别传给各自的组件,后面适配器只管工具,不管模型,链路就清晰了。
3. 可复制配置:LangChainTool 适配与链式封装
这一节是核心,给出三份可直接复制的配置:基础 LangChainTool 适配、批量工具集转换、以及把 LangChain Chain 封装成 crewAI 工具的完整代码。每份都标注了文件路径和关键参数。
先看基础适配。假设你有一个 LangChain 的 Wikipedia 工具,想塞进 crewAI Agent:
# tools/langchain_adapter.py from crewai.tools import LangChainTool from langchain_community.tools import WikipediaQueryRun from langchain_community.utilities import WikipediaAPIWrapper # 原始 LangChain 工具 wikipedia_lc = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper()) # 包装成 crewAI 工具,覆盖名称和描述以适配中文场景 wikipedia_tool = LangChainTool( tool=wikipedia_lc, name="百科知识查询", description="查询维基百科获取可靠的背景知识,输入为查询关键词字符串。", )name和description覆盖很重要。crewAI 的 Agent 是根据工具描述来决定调不调、怎么调的,如果沿用 LangChain 原始的英文描述,中文场景下 Agent 的调用决策会不稳定。描述里最好写清楚输入格式,比如"输入为查询关键词字符串",这样 Agent 生成的参数不会跑偏。
批量转换工具集,比如 SQLDatabaseToolkit 返回一组工具:
# tools/sql_toolkit_adapter.py from crewai.tools import LangChainTool from langchain_community.agent_toolkits import SQLDatabaseToolkit from langchain_community.utilities import SQLDatabase from langchain_openai import ChatOpenAI db = SQLDatabase.from_uri("postgresql://user:pass@localhost:5432/mydb") llm = ChatOpenAI(model="gpt-4o-mini", base_url="https://taotoken.net/api", temperature=0) sql_toolkit = SQLDatabaseToolkit(db=db, llm=llm) lc_tools = sql_toolkit.get_tools() # 批量包装,保留原始名称避免 Agent 混淆 crewai_tools = [LangChainTool(tool=t) for t in lc_tools]批量转换时不要覆盖名称,因为 SQL 工具集内部有调用顺序依赖(先 list_tables 再 query),名称改了 Agent 可能乱序调用。只有在单个工具、语义明确时才覆盖。
链式封装是重点。LangChain 的 LCEL 表达式或 RetrievalQA 本身不是BaseTool,不能直接喂给LangChainTool,需要用 crewAI 的BaseTool手动包一层。关键是把参数 schema 和返回值格式固定下来:
# tools/sentiment_chain_tool.py from crewai.tools import BaseTool from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from pydantic import BaseModel, Field def create_sentiment_chain(): llm = ChatOpenAI( model="gpt-4o-mini", base_url="https://taotoken.net/api", temperature=0, ) prompt = ChatPromptTemplate.from_template( "分析以下文本的情感倾向(正面/负面/中性)," "给出评分(-1到1)和理由:\n\n{text}" ) return prompt | llm | StrOutputParser() class SentimentInput(BaseModel): text: str = Field(description="需要进行情感分析的文本内容") class SentimentAnalysisTool(BaseTool): name: str = "情感分析工具" description: str = ( "分析文本的情感倾向,返回正面/负面/中性的判断和-1到1的评分。" "输入为待分析文本字符串。" ) args_schema: type[BaseModel] = SentimentInput def __init__(self, **kwargs): super().__init__(**kwargs) self._chain = create_sentiment_chain() def _run(self, text: str) -> str: return self._chain.invoke({"text": text})这里args_schema用 Pydantic 模型定义,字段名text必须和 Chain 里 prompt 的变量名{text}一致,否则invoke时会报 KeyError。返回值统一成字符串,因为 crewAI 工具的输出最终会拼进 Agent 的上下文,返回 dict 或 list 会导致序列化问题。如果你需要结构化返回,在_run里自己json.dumps成字符串。
更复杂的 RAG Chain 封装同理,只是_run里多一步来源提取:
# tools/rag_chain_tool.py import json from crewai.tools import BaseTool from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain.chains import RetrievalQA from pydantic import BaseModel, Field class RAGInput(BaseModel): question: str = Field(description="关于内部知识库的问题") class InternalKnowledgeRAGTool(BaseTool): name: str = "内部知识库查询工具" description: str = ( "在公司内部文档库中搜索答案,返回答案文本和来源列表。" "输入为问题字符串。仅用于查询公司内部信息。" ) args_schema: type[BaseModel] = RAGInput def __init__(self, vectorstore_path: str, **kwargs): super().__init__(**kwargs) embeddings = OpenAIEmbeddings( model="text-embedding-3-small", base_url="https://taotoken.net/api", ) vectorstore = Chroma( persist_directory=vectorstore_path, embedding_function=embeddings, ) self._qa_chain = RetrievalQA.from_chain_type( llm=ChatOpenAI( model="gpt-4o-mini", base_url="https://taotoken.net/api", temperature=0, ), retriever=vectorstore.as_retriever(search_kwargs={"k": 5}), return_source_documents=True, ) def _run(self, question: str) -> str: result = self._qa_chain({"query": question}) sources = list({ doc.metadata.get("source", "未知") for doc in result["source_documents"] }) return json.dumps( {"answer": result["result"], "sources": sources}, ensure_ascii=False, )注意_run返回的是 JSON 字符串而不是 dict,这样 Agent 拿到的是可读文本,同时下游工具如果需要解析也能json.loads。链式复用的关键就在这里:上游工具的输出格式固定,下游工具或 Agent 才能稳定消费。
4. 验证请求:跑通跨框架工具调用链路
配置写完必须验证,不然你不知道是适配器没生效还是模型没调通。我按从简到繁三步验证。
第一步,单独验证 LangChainTool 包装是否生效。不启动 Agent,直接调工具的run方法:
# verify_step1.py from tools.langchain_adapter import wikipedia_tool result = wikipedia_tool.run("crewAI multi-agent framework") print(type(result)) print(result[:200])如果打印出字符串且内容是百科摘要,说明适配器工作正常。如果报AttributeError: 'LangChainTool' object has no attribute 'run',检查 crewAI 版本,1.11.0 之前的方法名可能是_run。
第二步,验证 Agent 能自主调用工具。构造一个最小 crewAI Agent:
# verify_step2.py from crewai import Agent, Task, Crew, Process from tools.langchain_adapter import wikipedia_tool researcher = Agent( role="学术研究员", goal="基于维基百科回答技术问题", backstory="你擅长用百科资料快速给出准确背景。", tools=[wikipedia_tool], verbose=True, ) task = Task( description="查询 crewAI 是什么,用两句话总结。", expected_output="两句话的中文总结。", agent=researcher, ) crew = Crew( agents=[researcher], tasks=[task], process=Process.sequential, verbose=True, ) result = crew.kickoff() print(result)跑起来后看 verbose 输出,应该能看到 Agent 决定调用"百科知识查询"工具、传入查询词、拿到结果、再生成总结。如果 Agent 没调工具直接编答案,说明工具描述不够明确,回去改description,加上"当需要外部事实时优先调用本工具"。
第三步,验证链式复用。把情感分析工具和 RAG 工具串起来,让一个 Agent 先查知识库、再分析情感:
# verify_step3.py from crewai import Agent, Task, Crew, Process from tools.rag_chain_tool import InternalKnowledgeRAGTool from tools.sentiment_chain_tool import SentimentAnalysisTool rag_tool = InternalKnowledgeRAGTool(vectorstore_path="./knowledge") sentiment_tool = SentimentAnalysisTool() analyst = Agent( role="用户反馈分析师", goal="先查内部知识库了解产品背景,再分析用户评论情感", backstory="你擅长结合内部资料做情感分析。", tools=[rag_tool, sentiment_tool], verbose=True, ) task = Task( description=( "用户评论:'这个功能太慢了,但客服响应很快'。" "先查询知识库中关于性能优化的说明,再分析这条评论的情感。" ), expected_output="包含知识库引用和情感评分的中文分析。", agent=analyst, ) crew = Crew(agents=[analyst], tasks=[task], process=Process.sequential, verbose=True) print(crew.kickoff())成功的话,verbose 里会先出现 RAG 工具调用、返回带 sources 的 JSON 字符串,再出现情感分析工具调用、返回评分。两个工具的参数名分别是question和text,互不干扰,这就是参数一致性验证通过。返回值都是字符串,Agent 能连续消费,链式复用成立。
如果你想让多个 Agent 接力,把 RAG 结果作为下一个 Task 的输入,用context参数传递:
task1 = Task(description="查询知识库", expected_output="答案", agent=analyst) task2 = Task( description="基于上一个任务的答案做情感分析", expected_output="情感报告", agent=analyst, context=[task1], )5. 本篇常见错排查
跨框架适配的报错大多集中在几个固定位置,我按实际遇到的频率列出来。
401 Unauthorized / invalid api key。这个最常见,通常是环境变量没生效或两边配置不一致。先确认echo $OPENAI_API_KEY有值,再检查 LangChain 和 crewAI 是否都读到了同一个变量。如果你在代码里硬编码了 key,注意 crewAI 的LLM和 LangChain 的ChatOpenAI参数名不同,前者是api_key,后者也是api_key,但 base_url 前者叫base_url、后者也叫base_url,别写成api_base。TaoToken 的地址是https://taotoken.net/api,不要漏掉/api后缀,也不要多加/v1,LangChain 会自己拼。
local proxy failed / connection refused。这个报错说明请求根本没发出去,通常是 base_url 写错或本地网络配置问题。检查base_url是不是https://taotoken.net/api,注意是 https 不是 http。如果你在容器里跑,确认容器能访问外网。这个报错和适配器无关,是网络层问题,先把curl https://taotoken.net/api/v1/models -H "Authorization: Bearer $OPENAI_API_KEY"跑通再回来调代码。
Error reading choices / KeyError 'choices'。这个说明请求发出去了但响应格式不对,常见原因是模型 ID 写错,服务端返回了错误 JSON 而不是标准 completion。crewAI 侧模型要写openai/gpt-4o-mini,LangChain 侧写gpt-4o-mini,两边格式不同。如果你用了 TaoToken 上某个特定模型,先去模型对话页面确认 ID 拼写,复制粘贴不要手打。
OAuth / authentication_error。如果你之前配过其他 provider 的凭证,环境里可能残留了冲突的变量。检查env | grep -i openai和env | grep -i anthropic,把不相关的清掉。crewAI 有时会读ANTHROPIC_API_KEY,如果你同时设了它和OPENAI_API_KEY,Agent 可能走错 provider。统一用 OpenAI 协议接入时,只保留OPENAI_API_KEY和OPENAI_API_BASE。
args_schema validation error。链式封装时 Pydantic 字段名和 Chain 变量名不一致会报这个。比如SentimentInput里字段叫text,但 prompt 里写的是{content},invoke时就会失败。解决办法是让字段名、prompt 变量名、_run参数名三者完全一致。另外 Pydantic v2 里args_schema的类型标注要用type[BaseModel],写成BaseModel会警告。
Agent 不调用工具,直接编答案。这不是报错但很常见。原因是工具description太模糊,Agent 判断不需要调。把描述改具体,写清楚"什么时候必须调用"和"输入格式是什么"。比如"当问题涉及外部事实、需要可靠来源时,必须调用本工具,输入为查询关键词字符串"。
返回值类型漂移导致下游解析失败。链式复用里上游工具返回 dict、下游工具期望 str,就会在json.loads或字符串拼接时报错。统一约定:所有 crewAI 工具的_run返回字符串,需要结构化就在字符串里放 JSON。这样无论 Agent 还是下游工具,拿到的都是可预期的类型。
6. 语义一致 CTA 与后续接入
跑通上面的验证后,你手上应该有一套能工作的跨框架工具链路:LangChain 工具通过LangChainTool适配进 crewAI,LangChain Chain 通过BaseTool封装成 crewAI 工具,参数和返回值在链式复用中保持一致。接下来如果要把它用到实际项目,有几个方向可以继续。
如果你需要更多模型做对比测试,比如某些任务用gpt-4o-mini、某些用更强的模型,可以在模型对话页面直接试不同模型的效果,确认哪个适合你的工具调用场景,再写进配置。模型 ID 确认好之后,回到代码里改model字段即可,base_url 不用动。
如果你要把这套链路做成长期运行的编码或 Agent 服务,建议看一下 Coding Plan,它适合需要稳定调用、按量计费的场景,比每次手动配 key 省事。接入文档里有 LangChain、crewAI 以及其他框架的完整配置示例,包括环境变量、base_url、模型 ID 的对应关系,遇到配置问题可以直接对照。
API Key 在控制台的 API Keys 页面管理,建议给不同项目建不同的 key,方便排查问题时定位是哪个项目超了额度。如果你用 Claude Code 做开发,它的接入配置和本文的 OpenAI 协议略有不同,参考对应的接入文档,核心思路一样:统一 base_url、统一 key、模型 ID 按框架格式写。
最后提醒一个实操细节:跨框架适配的调试成本主要在版本和配置格式上,代码逻辑本身不复杂。把依赖版本卡死、把两边的模型配置对齐、把工具描述写清楚,这三件事做到位,后面基本不会出问题。我踩过的坑大多是因为版本混用和模型 ID 格式写错,跟适配器本身没关系。