Spaces:
Sleeping
Sleeping
File size: 2,932 Bytes
d7b3d84 | 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 | import asyncio
import http.client
import json
import os
import sys
sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
from dotenv import load_dotenv
load_dotenv()
import logging
from pydantic import BaseModel
from browser_use import ActionResult, Agent, ChatOpenAI, Tools
from browser_use.browser.profile import BrowserProfile
logger = logging.getLogger(__name__)
class Person(BaseModel):
name: str
email: str | None = None
class PersonList(BaseModel):
people: list[Person]
SERP_API_KEY = os.getenv('SERPER_API_KEY')
if not SERP_API_KEY:
raise ValueError('SERPER_API_KEY is not set')
tools = Tools(exclude_actions=['search'], output_model=PersonList)
@tools.registry.action('Search the web for a specific query. Returns a short description and links of the results.')
async def search_web(query: str):
# do a serp search for the query
conn = http.client.HTTPSConnection('google.serper.dev')
payload = json.dumps({'q': query})
headers = {'X-API-KEY': SERP_API_KEY, 'Content-Type': 'application/json'}
conn.request('POST', '/search', payload, headers)
res = conn.getresponse()
data = res.read()
serp_data = json.loads(data.decode('utf-8'))
# exclude searchParameters and credits
serp_data = {k: v for k, v in serp_data.items() if k not in ['searchParameters', 'credits']}
# keep the value of the key "organic"
organic = serp_data.get('organic', [])
# remove the key "position"
organic = [{k: v for k, v in d.items() if k != 'position'} for d in organic]
# print the original data
logger.debug(json.dumps(organic, indent=2))
# to string
organic_str = json.dumps(organic)
return ActionResult(extracted_content=organic_str, include_in_memory=False, include_extracted_content_only_once=True)
names = [
'Ruedi Aebersold',
'Bernd Bodenmiller',
'Eugene Demler',
'Erich Fischer',
'Pietro Gambardella',
'Matthias Huss',
'Reto Knutti',
'Maksym Kovalenko',
'Antonio Lanzavecchia',
'Maria Lukatskaya',
'Jochen Markard',
'Javier Pérez-Ramírez',
'Federica Sallusto',
'Gisbert Schneider',
'Sonia I. Seneviratne',
'Michael Siegrist',
'Johan Six',
'Tanja Stadler',
'Shinichi Sunagawa',
'Michael Bruce Zimmermann',
]
async def main():
task = 'use search_web with "find email address of the following ETH professor:" for each of the following persons in a list of actions. Finally return the list with name and email if provided - do always 5 at once'
task += '\n' + '\n'.join(names)
model = ChatOpenAI(model='gpt-4.1-mini')
browser_profile = BrowserProfile()
agent = Agent(task=task, llm=model, tools=tools, browser_profile=browser_profile)
history = await agent.run()
result = history.final_result()
if result:
parsed: PersonList = PersonList.model_validate_json(result)
for person in parsed.people:
print(f'{person.name} - {person.email}')
else:
print('No result')
if __name__ == '__main__':
asyncio.run(main())
|