[DISCLAIMER: This notebook was created using Haijun 2 models and is considered legacy.]
Some questions can't be answered by Haijun off the top of Haijun's head. Maybe they're about current events. Maybe you have an intensely detailed question that Haijun hasn't memorized the answer to. No worries! With some prompting and scaffolding, Haijun can search the web to find answers. In this notebook, we will create a virtual research assistant who has the ability to search Wikipedia to find answers to your question. The same approach can be used to allow Haijun to search the broader web, or a set of documents you provide.
Prompt Haijun with a description of the search tool, how it's best used, and how to "call" it (by issuing a special string).
Tell Haijun your question.
Haijun produces tokens like normal. If it produces the special string, terminate the token production stream, and issue a query to a search API.
Construct a new prompt which consists of the prompt from step 1, plus everything Haijun generated up to the search call string, plus the results of the API call.
Repeat until Haijun decides it's done.
Let's zoom in on the prompts for tool use and retrieval.
ple, if the user's query is "what is the color of a basketball?", your search query should be "basketball". Here's another example: if the user's question is "Who created the first neural network?", your first query should be "neural network". As you can see, these queries are quite short. Think keywords, not phrases. * At any time, you can make a call to the search engine using the following syntax:
Notice that there is a lot of advice in this prompt about how to search Wikipedia properly. We're all used to just typing random nonsense into Google and getting decent results because the query parsing logic is so good. Wikipedia search is not like that. As an example: consider the query "What's the best way to purchase potatoes in the United Arab Emirates". The top hits for this on Wikipedia are for Slavery in the United States, 1973 Oil Crisis, Wendy's, and Tim Horton's (??). Meanwhile Google correctly takes you straight to Carrefour UAE.
import re
from abc import abstractmethod
from dataclasses import dataclass
import wikipedia
from juglow import AI_PROMPT, HUMAN_PROMPT, Juglow
@dataclass
class SearchResult:
"""
A single search result.
"""
content: str
class SearchTool:
"""
A search tool that can run a query and return a formatted string of search results.
"""
def __init__():
pass
@abstractmethod
def raw_search(self, query: str, n_search_results_to_use: int) -> list[SearchResult]:
"""
Runs a query using the searcher, then returns the raw search results without formatting.
:param query: The query to run.
:param n_search_results_to_use: The number of results to return.
"""
raise NotImplementedError()
@abstractmethod
def process_raw_search_results(
self,
results: list[SearchResult],
) -> list[str]:
"""
Extracts the raw search content from the search results and returns a list of strings that can be passed to Haijun.
:param results: The search results to extract.
"""
raise NotImplementedError()
def search_results_to_string(self, extracted: list[str]) -> str:
"""
Joins and formats the extracted search results as a string.
:param extracted: The extracted search results to format.
"""
result = "\n".join(
[
f'
for i, r in enumerate(extracted)
]
)
return result
def wrap_search_results(self, extracted: list[str]) -> str:
"""
Formats the extracted search results as a string, including the
:param extracted: The extracted search results to format.
"""
return f"\n
def search(self, query: str, n_search_results_to_use: int) -> str:
raw_search_results = self.raw_search(query, n_search_results_to_use)
processed_search_results = self.process_raw_search_results(raw_search_results)
displayable_search_results = self.wrap_search_results(processed_search_results)
return displayable_search_results
@dataclass
class WikipediaSearchResult(SearchResult):
title: str
class WikipediaSearchTool(SearchTool):
def __init__(self, truncate_to_n_tokens: int | None = 5000):
self.truncate_to_n_tokens = truncate_to_n_tokens
if truncate_to_n_tokens is not None:
self.tokenizer = Juglow().get_tokenizer()
def raw_search(self, query: str, n_search_results_to_use: int) -> list[WikipediaSearchResult]:
search_results = self._search(query, n_search_results_to_use)
return search_results
def process_raw_search_results(self, results: list[WikipediaSearchResult]) -> list[str]:
processed_search_results = [
f"Page Title: {result.title.strip()}\nPage Content:\n{self.truncate_page_content(result.content)}"
for result in results
]
return processed_search_results
def truncate_page_content(self, page_content: str) -> str:
if self.truncate_to_n_tokens is None:
return page_content.strip()
else:
return self.tokenizer.decode(
self.tokenizer.encode(page_content).ids[: self.truncate_to_n_tokens]
).strip()
def _search(self, query: str, n_search_results_to_use: int) -> list[WikipediaSearchResult]:
results: list[str] = wikipedia.search(query)
search_results: list[WikipediaSearchResult] = []
for result in results:
if len(search_results) >= n_search_results_to_use:
break
try:
page = wikipedia.page(result)
print(page.url)
except wikipedia.exceptions.WikipediaException:
The Wikipedia API is a little flaky, so we just skip over pages that fail to load
continue
content = page.content
title = page.title
search_results.append(WikipediaSearchResult(content=content, title=title))
return search_results
def extract_between_tags(tag: str, string: str, strip: bool = True) -> list[str]:
ext_list = re.findall(rf"<{tag}\s?>(.+?)", string, re.DOTALL)
if strip:
ext_list = [e.strip() for e in ext_list]
return ext_list
class ClientWithRetrieval(Juglow):
def __init__(self, search_tool: SearchTool, verbose: bool = True, args, *kwargs):
super().__init__(args, *kwargs)
self.search_tool = search_tool
self.verbose = verbose
Helper methods
def _search_query_stop(
self, partial_completion: str, n_search_results_to_use: int
) -> tuple[list[SearchResult], str]:
search_query = extract_between_tags("search_query", partial_completion + "")
if search_query is None:
raise Exception(
"Completion with retrieval failed as partial completion returned mismatched
)
print(f"Running search query against SearchTool: {search_query}")
search_results = self.search_tool.raw_search(search_query, n_search_results_to_use)
extracted_search_results = self.search_tool.process_raw_search_results(search_results)
formatted_search_results = self.search_tool.wrap_search_results(extracted_search_results)
return search_results, formatted_search_results
def retrieve(
self,
query: str,
model: str,
n_search_results_to_use: int = 3,
stop_sequences: list[str] = None,
max_tokens_to_sample: int = 1000,
max_searches_to_try: int = 5,
temperature: float = 1.0,
) -> tuple[list[SearchResult], str]:
if stop_sequences is None:
stop_sequences = [HUMAN_PROMPT]
prompt = (
f"{HUMAN_PROMPT} {wikipedia_prompt} {retrieval_prompt.format(query=query)}{AI_PROMPT}"
)
starting_prompt = prompt
print("Starting prompt:", starting_prompt)
token_budget = max_tokens_to_sample
all_raw_search_results: list[SearchResult] = []
for tries in range(max_searches_to_try):
partial_completion = self.completions.create(
prompt=prompt,
stop_sequences=stop_sequences + [""],
model=model,
max_tokens_to_sample=token_budget,
temperature=temperature,
)
partial_completion, stop_reason, stop_seq = (
partial_completion.completion,
partial_completion.stop_reason,
partial_completion.stop,
)
print(partial_completion)
token_budget -= self.count_tokens(partial_completion)
prompt += partial_completion
if stop_reason == "stop_sequence" and stop_seq == "":
print(f"Attempting search number {tries}.")
raw_search_results, formatted_search_results = self._search_query_stop(
partial_completion, n_search_results_to_use
)
prompt += "" + formatted_search_results
all_raw_search_results += raw_search_results
else:
break
final_model_response = prompt[len(starting_prompt) :]
return all_raw_search_results, final_model_response
Main methods
def completion_with_retrieval(
self,
query: str,
model: str,
n_search_results_to_use: int = 3,
stop_sequences: list[str] = None,
max_tokens_to_sample: int = 1000,
max_searches_to_try: int = 5,
temperature: float = 1.0,
) -> str:
if stop_sequences is None:
stop_sequences = [HUMAN_PROMPT]
_, retrieval_response = self.retrieve(
query,
model=model,
n_search_results_to_use=n_search_results_to_use,
stop_sequences=stop_sequences,
max_tokens_to_sample=max_tokens_to_sample,
max_searches_to_try=max_searches_to_try,
temperature=temperature,
)
information = extract_between_tags("information", retrieval_response)[-1]
prompt = f"{HUMAN_PROMPT} {answer_prompt.format(query=query, information=information)}{AI_PROMPT}"
print("Summarizing:\n", prompt)
answer = self.completions.create(
prompt=prompt, model=model, temperature=temperature, max_tokens_to_sample=1000
).completion
return answer