Skip to content

第三方工具集成

ADK 被设计为高度可扩展的系统，可无缝集成来自其他 AI Agent 框架（如 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 密钥并设置为环境变量：

    bashpy 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")

```py
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?")