Skip to content

第三方工具集成

ADK 设计为高度可扩展,允许你无缝集成来自其他 AI 智能体框架(如 CrewAI 和 LangChain)的工具。这种互操作性至关重要,因为它可以缩短开发时间并允许你重用现有工具。

1. 使用 LangChain 工具

ADK 提供了 LangchainTool 包装器,可将 LangChain 生态系统中的工具集成到你的智能体中。

示例:使用 LangChain 的 Tavily 工具进行网络搜索

Tavily 提供了一个搜索 API,返回从实时搜索结果中获取的答案,供 AI 智能体等应用程序使用。

  1. 遵循 ADK 安装和设置 指南。

  2. 安装依赖项: 确保你已安装必要的 LangChain 包。例如,要使用 Tavily 搜索工具,安装其特定依赖项:

    pip install langchain_community tavily-python
    
  3. 获取 Tavily API KEY 并将其导出为环境变量。

    export TAVILY_API_KEY=<REPLACE_WITH_API_KEY>
    
  4. 导入: 从 ADK 导入 LangchainTool 包装器和你希望使用的特定 LangChain 工具(例如,TavilySearchResults)。

    from google.adk.tools.langchain_tool import LangchainTool
    from langchain_community.tools import TavilySearchResults
    
  5. 实例化与包装: 创建 LangChain 工具的实例并将其传递给 LangchainTool 构造函数。

    # 实例化 LangChain 工具
    tavily_tool_instance = TavilySearchResults(
        max_results=5,
        search_depth="advanced",
        include_answer=True,
        include_raw_content=True,
        include_images=True,
    )
    
    # 使用 LangchainTool 为 ADK 包装它
    adk_tavily_tool = LangchainTool(tool=tavily_tool_instance)
    
  6. 添加到智能体: 在定义时将包装的 LangchainTool 实例包含在你的智能体的 tools 列表中。

    from google.adk import Agent
    
    # 定义 ADK 智能体,包括包装的工具
    my_agent = Agent(
        name="langchain_tool_agent",
        model="gemini-2.0-flash",
        description="Agent to answer questions using TavilySearch.",
        instruction="I can answer your questions by searching the internet. Just ask me anything!",
        tools=[adk_tavily_tool] # 在这里添加包装的工具
    )
    

这是完整代码,结合上述步骤创建并运行使用 LangChain Tavily 搜索工具的智能体。

import os
from google.adk import Agent, Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools.langchain_tool import LangchainTool
from google.genai import types
from langchain_community.tools import TavilySearchResults

# Ensure TAVILY_API_KEY is set in your environment
if not os.getenv("TAVILY_API_KEY"):
    print("Warning: TAVILY_API_KEY environment variable not set.")

APP_NAME = "news_app"
USER_ID = "1234"
SESSION_ID = "session1234"

# Instantiate LangChain tool
tavily_search = TavilySearchResults(
    max_results=5,
    search_depth="advanced",
    include_answer=True,
    include_raw_content=True,
    include_images=True,
)

# Wrap with LangchainTool
adk_tavily_tool = LangchainTool(tool=tavily_search)

# Define Agent with the wrapped tool
my_agent = Agent(
    name="langchain_tool_agent",
    model="gemini-2.0-flash",
    description="Agent to answer questions using TavilySearch.",
    instruction="I can answer your questions by searching the internet. Just ask me anything!",
    tools=[adk_tavily_tool] # Add the wrapped tool here
)

session_service = InMemorySessionService()
session = session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
runner = Runner(agent=my_agent, app_name=APP_NAME, session_service=session_service)


# Agent Interaction
def call_agent(query):
    content = types.Content(role='user', parts=[types.Part(text=query)])
    events = runner.run(user_id=USER_ID, session_id=SESSION_ID, new_message=content)

    for event in events:
        if event.is_final_response():
            final_response = event.content.parts[0].text
            print("Agent Response: ", final_response)

call_agent("stock price of GOOG")

2. 使用 CrewAI 工具

ADK 提供了 CrewaiTool 包装器,可集成来自 CrewAI 库的工具。

示例:使用 CrewAI 的 Serper API 进行网络搜索

Serper API 以编程方式提供对 Google 搜索结果的访问。它允许应用程序(如 AI 智能体)执行实时 Google 搜索(包括新闻、图像等)并获取结构化数据,而无需直接抓取网页。

  1. 遵循 ADK 安装和设置 指南。

  2. 安装依赖项: 安装必要的 CrewAI 工具包。例如,要使用 SerperDevTool:

    pip install crewai-tools
    
  3. 获取 Serper API KEY 并将其导出为环境变量。

    export SERPER_API_KEY=<REPLACE_WITH_API_KEY>
    
  4. 导入: 从 ADK 导入 CrewaiTool 和所需的 CrewAI 工具(例如,SerperDevTool)。

    from google.adk.tools.crewai_tool import CrewaiTool
    from crewai_tools import SerperDevTool
    
  5. 实例化与包装: 创建 CrewAI 工具的实例。将其传递给 CrewaiTool 构造函数。至关重要的是,你必须为 ADK 包装器提供名称和描述,因为这些被 ADK 的底层模型用来了解何时使用该工具。

    # 实例化 CrewAI 工具
    serper_tool_instance = SerperDevTool(
        n_results=10,
        save_file=False,
        search_type="news",
    )
    
    # 使用 CrewaiTool 为 ADK 包装它,提供名称和描述
    adk_serper_tool = CrewaiTool(
        name="InternetNewsSearch",
        description="Searches the internet specifically for recent news articles using Serper.",
        tool=serper_tool_instance
    )
    
  6. 添加到智能体: 将包装的 CrewaiTool 实例包含在你的智能体的 tools 列表中。

    from google.adk import Agent
    
    # 定义 ADK 智能体
    my_agent = Agent(
        name="crewai_search_agent",
        model="gemini-2.0-flash",
        description="Agent to find recent news using the Serper search tool.",
        instruction="I can find the latest news for you. What topic are you interested in?",
        tools=[adk_serper_tool] # 在这里添加包装的工具
    )
    

完整示例:Serper API

这是完整代码,结合上述步骤创建并运行使用 CrewAI Serper API 搜索工具的智能体。

import os
from google.adk import Agent, Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools.crewai_tool import CrewaiTool
from google.genai import types
from crewai_tools import SerperDevTool


# Constants
APP_NAME = "news_app"
USER_ID = "user1234"
SESSION_ID = "1234"

# Ensure SERPER_API_KEY is set in your environment
if not os.getenv("SERPER_API_KEY"):
    print("Warning: SERPER_API_KEY environment variable not set.")

serper_tool_instance = SerperDevTool(
    n_results=10,
    save_file=False,
    search_type="news",
)

adk_serper_tool = CrewaiTool(
    name="InternetNewsSearch",
    description="Searches the internet specifically for recent news articles using Serper.",
    tool=serper_tool_instance
)

serper_agent = Agent(
    name="basic_search_agent",
    model="gemini-2.0-flash",
    description="Agent to answer questions using Google Search.",
    instruction="I can answer your questions by searching the internet. Just ask me anything!",
    # Add the Serper tool
    tools=[adk_serper_tool]
)

# Session and Runner
session_service = InMemorySessionService()
session = session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
runner = Runner(agent=serper_agent, app_name=APP_NAME, session_service=session_service)


# Agent Interaction
def call_agent(query):
    content = types.Content(role='user', parts=[types.Part(text=query)])
    events = runner.run(user_id=USER_ID, session_id=SESSION_ID, new_message=content)

    for event in events:
        if event.is_final_response():
            final_response = event.content.parts[0].text
            print("Agent Response: ", final_response)

call_agent("what's the latest news on AI Agents?")