与外部 API 集成

Agents & External APIs

Tools for External Access

API: Your Agent's Interface

Securing API Access

Python & HTTP Requests

import requests

def fetch_joke():
    url = "https://official-joke-api.appspot.com/random_joke"
    response = requests.get(url)
    if response.status_code == 200:
        return response.json()
    else:
        return {"error": "Could not fetch joke."}

if __name__ == "__main__":
    joke_data = fetch_joke()
    if "error" not in joke_data:
        print(f"Setup: {joke_data['setup']}")
        print(f"Punchline: {joke_data['punchline']}")
    else:
        print(joke_data["error"])

From Function to LangChain Tool

from langchain.tools import BaseTool
from pydantic import BaseModel, Field
import requests

# Define the input schema for the tool
class JokeInput(BaseModel):
    # Our joke API doesn't need specific input,
    # but tools usually define what they expect.
    query: str = Field(description="A placeholder query, not used by this API.")

class JokeTool(BaseTool):
    name = "get_random_joke"
    description = (
        "Useful for when you need a random joke. "
        "Returns a setup and a punchline."
    )
    args_schema: type[BaseModel] = JokeInput

    def _run(self, query: str):
        url = "https://official-joke-api.appspot.com/random_joke"
        response = requests.get(url)
        if response.status_code == 200:
            joke = response.json()
            return f"Setup: {joke['setup']}\nPunchline: {joke['punchline']}"
        return "Failed to fetch a joke."

    async def _arun(self, query: str):
        # Asynchronous version (optional, but good practice)
        raise NotImplementedError("Asynchronous call not implemented for this tool yet.")

if __name__ == "__main__":
    joke_tool = JokeTool()
    print(joke_tool._run("tell me a joke"))

Agent with API Tool

# Assuming you have an OpenAI API key set as an environment variable
# export OPENAI_API_KEY="..."

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain import hub
from langchain.tools import BaseTool
from pydantic import BaseModel, Field
import requests
import os

# Re-define JokeTool for completeness in runnable snippet
class JokeInput(BaseModel):
    query: str = Field(description="A placeholder query, not used by this API.")

class JokeTool(BaseTool):
    name = "get_random_joke"
    description = "Useful for when you need a random joke. Returns a setup and a punchline."
    args_schema: type[BaseModel] = JokeInput

    def _run(self, query: str):
        url = "https://official-joke-api.appspot.com/random_joke"
        response = requests.get(url)
        if response.status_code == 200:
            joke = response.json()
            return f"Setup: {joke['setup']}\nPunchline: {joke['punchline']}"
        return "Failed to fetch a joke."

    async def _arun(self, query: str):
        raise NotImplementedError("Async not implemented.")

if __name__ == "__main__":
    # Ensure OPENAI_API_KEY is set in environment
    if not os.getenv("OPENAI_API_KEY"):
        print("Please set your OPENAI_API_KEY environment variable.")
        exit()

    llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
    tools = [JokeTool()]

    # Get the prompt for the ReAct agent
    prompt = hub.pull("hwchase17/react")

    agent = create_react_agent(llm, tools, prompt)
    agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

    print("Agent is ready!")
    # Example usage will be in the next scene.

Agent Using the API Tool

# This code block assumes the setup from the previous scene
# has been executed and the agent_executor is available.
# In a real interactive environment, you'd run this after Scene 7.

# For demonstration, we re-initialize parts needed for execution
import os
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain import hub
from langchain.tools import BaseTool
from pydantic import BaseModel, Field
import requests

# Re-define JokeTool for completeness
class JokeInput(BaseModel):
    query: str = Field(description="A placeholder query, not used by this API.")

class JokeTool(BaseTool):
    name = "get_random_joke"
    description = "Useful for when you need a random joke. Returns a setup and a punchline."
    args_schema: type[BaseModel] = JokeInput

    def _run(self, query: str):
        url = "https://official-joke-api.appspot.com/random_joke"
        response = requests.get(url)
        if response.status_code == 200:
            joke = response.json()
            return f"Setup: {joke['setup']}\nPunchline: {joke['punchline']}"
        return "Failed to fetch a joke."

    async def _arun(self, query: str):
        raise NotImplementedError("Async not implemented.")

if __name__ == "__main__":
    if not os.getenv("OPENAI_API_KEY"):
        print("Please set your OPENAI_API_KEY environment variable.")
        exit()

    llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
    tools = [JokeTool()]
    prompt = hub.pull("hwchase17/react")
    agent = create_react_agent(llm, tools, prompt)
    agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

    print("--- Asking the agent for a joke ---")
    response = agent_executor.invoke({"input": "Tell me a random joke."})
    print("\nAgent's final answer:")
    print(response["output"])

Robust API Integration

Beyond Simple APIs

API Integration Check

Recap & Next Steps