Build a Google Search Agent

📘
Welcome to the guide to build your first Agent!
This program allows you to ask questions and get answers using the powerful GPT-3.5 Turbo language model. If the Turbo model feels like it needs more information from a Google Search, it will tell us the query it would like to search Google for. We will the search using custom tool we've made, feed the result back to the Agent and continue until the Agent feels like it has enough information to give an accurate response.

Interactive Code Walkthrough

Click "Open Recipe" below to open a step-by-step code walkthrough

🦉

Agent with Google Search

Open Recipe

SGP APIs used in this guide

Execute Agent API

Prerequisites

Prerequisite	Description	Instructions
API Key	Required for SGP API requests	Follow instructions from the API Key Authentication Section
Serp API Key	Required for Google Search requests	Register for a free Serp API account at https://serpapi.com/

What You've Built

After completing the guide, you will be able to interact with the Agent using a CLI chat interface. This agent will be able to use Google Search to fill in information it does not know, otherwise it will respond directly to the user.

Full Code Snippet

import argparse
import requests
import readline
import json
from typing import Dict, List, Optional


# Replace this with your Spellbook API Key
SPELLBOOK_API_KEY = "sampleapikey12345"

# Replace this with your Serp API Key
SERP_API_KEY = "sampleapikey12345"



def post_request(url: str, data: Dict, api_key: Optional[str] = None):
    headers = {
        'accept': 'application/json',
        'Content-Type': 'application/json',
    }
    if api_key:
        headers['x-api-key'] = api_key
        
    response = requests.post(url, headers=headers, json=data)
    if response.status_code == 200:
        json_response = response.json()
        return json_response
    
    raise ValueError(f"Request failed ({response}): {response.text}")


def get_request(url: str, params: Dict, api_key: Optional[str] = None):
    headers = {
        'accept': 'application/json',
        'Content-Type': 'application/json',
    }
    if api_key:
        headers['x-api-key'] = api_key
        
    response = requests.get(url, headers=headers, params=params)
    if response.status_code == 200:
        json_response = response.json()
        return json_response
    
    raise ValueError(f"Request failed ({response}): {response.text}")


def search_google(query: str, limit: int = 10) -> str:
    """Searches google for a given query"""
    results = get_request(
        url="https://serpapi.com/search",
        params={
            "engine": "google",
            "q": query,
            "api_key": SERP_API_KEY,
        }
    )

    # Taken from Langchain SERP API wrapper helpers (_parse_snippets)
    # "places" and "images" is available from Serper but not implemented in the
    # parser of run(). They can be used in results()
    type: str = "search"
    result_key_for_type = {
        "news": "news",
        "places": "places",
        "images": "images",
        "search": "organic_results",
    }

    snippets = []

    if results.get("answerBox"):
        answer_box = results.get("answerBox", {})
        if answer_box.get("answer"):
            return [answer_box.get("answer")]
        elif answer_box.get("snippet"):
            return [answer_box.get("snippet").replace("\n", " ")]
        elif answer_box.get("snippetHighlighted"):
            return answer_box.get("snippetHighlighted")

    if results.get("knowledgeGraph"):
        kg = results.get("knowledgeGraph", {})
        title = kg.get("title")
        entity_type = kg.get("type")
        if entity_type:
            snippets.append(f"{title}: {entity_type}.")
        description = kg.get("description")
        if description:
            snippets.append(description)
        for attribute, value in kg.get("attributes", {}).items():
            snippets.append(f"{title} {attribute}: {value}.")

    for result in results[result_key_for_type[type]][: limit]:
        if "snippet" in result:
            snippets.append(result["snippet"])
        for attribute, value in result.get("attributes", {}).items():
            snippets.append(f"{attribute}: {value}.")

    if len(snippets) == 0:
        snippets = ["No good Google Search Result was found"]

    return " ".join(snippets)


def execute_agent(messages):
    data = {
        "memory_strategy": {
            "name": "last_k",
            "params": {
                "k": 20,
            }
        },
        "messages": messages,
        "model": "gpt-3.5-turbo-0613",
        "tools": [
            {
                "name": "search_google",
                "description": "A Google Search tool. Useful for when you need to answer questions about current events or about something you don't know.",
                "arguments": {
                    "type": "object",
                    "properties": {
                        "query": {
                            "type": "string",
                            "description": "The query to search for"
                        },
                    },
                    "required": ["query"]
                }
            },
        ],
    }
    return post_request(
        "https://api.spellbook.scale.com/egp/v1/agents/execute", 
        data, 
        api_key=SPELLBOOK_API_KEY
    )


def run(messages: List, max_turns: int = 10):
    agent_response = None
    remaining_turns = max_turns
    
    # Repeatedly call the agent API on the updated list of messages until the agent returns
    # final content or the agent exceeds its max_turns limit.
    while not agent_response or agent_response['action'] != 'content' or remaining_turns <= 0:
        agent_response = execute_agent(messages)
        if agent_response['action'] == 'content':
            # Append the final agent response to the messages
            messages.append({"role": "agent", "content": agent_response['context']['content']})
            
        elif agent_response['action'] == 'tool_request':
            # Append the agent's tool request to the messages
            messages.append({"role": "agent", "tool_request": agent_response['context']['tool_request']})
            
            # Extract the tool name and args from the agent response
            tool_name = agent_response['context']['tool_request']['name']
            tool_args = json.loads(agent_response['context']['tool_request']['arguments'])
            
            # Execute the chosen tool and extract the response
            if tool_name == 'search_google':
                tool_response = search_google(tool_args['query'])
            else:
                raise ValueError(f"Unknown tool: {tool_name}")
                
            # Append the tool response to the messages
            messages.append({"role": "tool", "name": tool_name, "content": tool_response})
        else:
            raise ValueError(f"Unknown action: {agent_response['action']}")
        remaining_turns -= 1
    
    return messages

  
def format_message(message):
    return ", ".join([f"{key.title()}: {value}" for key, value in message.items()])


if __name__ == "__main__":
    argparser = argparse.ArgumentParser()
    argparser.add_argument("-v", "--verbose", action="store_true", help="Print the message history")
    args = argparser.parse_args()

    messages = []
    while True:
        print("\n" + "=" * 50 + "\n")
        user_input = input("Ask me a question: ")
        messages.append({"role": "user", "content": user_input})
        new_messages = run(messages)
        if args.verbose:
            print("\nMessage History:")
            for message in new_messages:
                print(format_message(message) + "\n")

            print("-" * 50 + "\n")
            print("\nQuestion:\n" + user_input)
        print("\nAnswer:\n" + new_messages[-1]['content'] + "\n")

📘Welcome to the guide to build your first Agent!

Interactive Code Walkthrough

SGP APIs used in this guide

Prerequisites

What You've Built

Full Code Snippet

📘
Welcome to the guide to build your first Agent!