Spaces:
Runtime error
Runtime error
File size: 6,657 Bytes
5831cdb 7118dfb fa2543e a7abe3e fa2543e 7118dfb 0189767 fa2543e 7b69047 fa2543e 5831cdb 7b69047 5831cdb af5c38f 5831cdb 85f69d5 5831cdb 7118dfb ae3368c 7118dfb 5c5bd6b 7118dfb 5831cdb 4a81f80 5c5bd6b 7118dfb 85f69d5 3a6ff6b 85f69d5 3a6ff6b 1e77711 3a6ff6b 5831cdb 7118dfb 0189767 ae3368c 7b69047 0c429cb 0189767 0c429cb 0189767 fa2543e 7b69047 ae3368c fa2543e a7abe3e ae3368c a7abe3e 115169a 89578da 115169a a7abe3e 89578da fa2543e 1e77711 115169a fa2543e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 |
import gplace
from typing import TypedDict, Optional
from langchain_google_community import GoogleSearchAPIWrapper
import utils
## Document vector store for context
from langchain_chroma import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import CSVLoader
from langchain_openai import OpenAIEmbeddings
import glob
utils.load_env()
search = GoogleSearchAPIWrapper()
class NearbySearchInput(TypedDict):
keyword: str
location_name: str
radius: int
place_type: Optional[str]
class NearbyDenseCommunityInput(TypedDict):
location_name: str
radius: int
# class GoogleSearchInput(TypedDict):
# keyword: str
# %%
def find_place_from_text(location:str):
"""Finds a place location and related data from the query text"""
print("function call find_place_from_text", location)
result = gplace.find_place_from_text(location)
r = result['candidates'][0]
return f"""
address: {r['formatted_address']}\n
location: {r['geometry']['location']}\n
location_name: {r['name']}\n
"""
# return f"""
# address: {r['formatted_address']}\n
# location: {r['geometry']['location']}\n
# location_name: {r['name']}\n
# """
# def nearby_search(keyword:str, location:str, radius=2000, place_type=None):
# """Searches for many places nearby the location based on a keyword. using keyword like \"coffee shop\", \"restaurants\". radius is the range to search from the location"""
# location = gplace.find_location(location, radius=radius)
# result = gplace.nearby_search(keyword, location, radius)
# strout = ""
# for r in result:
# strout = strout + f"""
# address: {r['vicinity']}\n
# location: {r['geometry']['location']}\n
# name: {r['name']}\n
# opening hours: {r['opening_hours']}\n
# rating: {r['rating']}\n
# plus code: {r['plus_code']['global_code']}\n\n
# """
# return strout
def nearby_search(input_dict: NearbySearchInput):
"""Searches for many places nearby the location based on a keyword. using keyword like \"coffee shop\", \"restaurants\". radius is the range to search from the location."""
print("function call nearby_search", input_dict, "\n")
max_results = 10
keyword = input_dict['keyword']
location = input_dict['location_name']
radius = input_dict.get('radius', 2000)
place_type = input_dict.get('place_type', None)
# Call the internal function to find location
location_coords = gplace.find_location(location, radius=radius)
result = gplace.nearby_search(keyword, location_coords, radius)
number_results = len(result)
strout = "number of results more than {}\n".format(number_results) if number_results==60 else "number of results: {}\n".format(number_results)
for r in result[:max_results]:
# Use .get() to handle missing keys
address = r.get('vicinity', 'N/A')
location_info = r.get('geometry', {}).get('location', 'N/A')
name = r.get('name', 'N/A')
opening_hours = r.get('opening_hours', 'N/A')
rating = r.get('rating', 'N/A')
plus_code = r.get('plus_code', {}).get('global_code', 'N/A')
# strout += f"""
# address: {address}\n
# location: {location_info}\n
# lacation_name: {name}\n
# opening hours: {opening_hours}\n
# rating: {rating}\n
# plus code: {plus_code}\n\n
# """
strout += f"""
**{name}**\n
address: {address}\n
rating: {rating}\n\n
"""
return strout
def nearby_dense_community(input_dict: NearbyDenseCommunityInput) -> str:
""" getting nearby dense community such as (community mall, hotel, school, etc), by location name, radius(in meters)
return list of location community nearby, name, community type.
"""
print("function call nearby_dense_community", input_dict, "\n")
max_results = 5
location = input_dict['location_name']
radius = input_dict['radius']
location_coords = gplace.find_location(location, radius=radius)
result = gplace.nearby_dense_community(location_coords, radius)
strout = ""
for r in result[:max_results]:
# Use .get() to handle missing keys
address = r.get('vicinity', 'N/A')
location_types = r.get('types', 'N/A')
name = r.get('name', 'N/A')
opening_hours = r.get('opening_hours', 'N/A')
rating = r.get('rating', 'N/A')
plus_code = r.get('plus_code', {}).get('global_code', 'N/A')
strout += f"""
name: {name}\n
types: {location_types}\n
"""
return strout
def google_search(keyword:str):
"""Search Google for recent results. Using keyword as a text query search in google."""
print("function call google_search", keyword, "\n")
return search.run(keyword).replace("\U000f1676", "")
## Document csv
def get_documents(file_pattern="document/*.csv"):
file_paths = tuple(glob.glob(file_pattern))
all_docs = []
for file_path in file_paths:
loader = CSVLoader(file_path=file_path)
docs = loader.load()
all_docs.extend(docs) # Add the documents to the list
return all_docs
def get_retriver_from_docs(docs):
print("function call get_retriver_from_docs", docs, "\n")
# Split text into chunks separated.
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=100)
splits = text_splitter.split_documents(docs)
# Text Vectorization.
vectorstore = Chroma.from_documents(documents=splits, embedding=OpenAIEmbeddings())
# Retrieve and generate using the relevant snippets of the blog.
retriever = vectorstore.as_retriever()
return retriever
from langchain.tools.retriever import create_retriever_tool
from langchain_core.tools import tool
from langchain_core.tools import Tool
docs = get_documents()
retriever = get_retriver_from_docs(docs)
population_doc_retriever = create_retriever_tool(
retriever,
"search_population_community_household_expenditures_data",
"Use this tool to retrieve information about population, community and household expenditures. by searching distinct or province"
)
# google_search = Tool(
# name="google_search",
# description="Search Google for recent results.",
# func=search.run,
# )
google_search = tool(google_search)
find_place_from_text = tool(find_place_from_text)
nearby_search = tool(nearby_search)
nearby_dense_community = tool(nearby_dense_community) |