Spaces:
Sleeping
Sleeping
File size: 2,518 Bytes
eb7faf3 36d6e26 eb7faf3 36d6e26 eb7faf3 36d6e26 eb7faf3 36d6e26 eb7faf3 | 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 | from agents import build_reader_agent, build_search_agent, writer_chain, critic_chain
import time
def extract_text(content) -> str:
if isinstance(content, str):
return content
elif isinstance(content, list):
return "".join(item.get("text", "") for item in content if isinstance(item, dict) and "text" in item)
return str(content)
def run_research_pipeline(topic: str) -> dict:
state={}
#Search Agent working
print("\n" + "="*50 )
print("step 1 - search agent is wokring ...")
print("=" *50)
search_agent= build_search_agent()
search_result= search_agent.invoke({
"messages": [("user", f"Find recent, reliable and detailed information about: {topic}")]
})
state["search_results"]= extract_text(search_result['messages'][-1].content)
print("\n search result", state['search_results'])
# Introduce delay to prevent rate limits
time.sleep(5)
#Step 2 - reader agent
print("\n" + "="*50)
print("step 2- reader agent is scrapping top respurces ...")
print("="*50)
reader_agent= build_reader_agent()
reader_result = reader_agent.invoke({
"messages": [("user",
f"Based on the following search results about '{topic}',"
f"pick the most relevant URL and scrape it for deeper content.\n\n"
f"Search Results: \n{state['search_results'][:800]}"
)]
})
state['scraped_content']= extract_text(reader_result['messages'][-1].content)
print("\nScraped content\n", state['scraped_content'])
# Introduce delay to prevent rate limits
time.sleep(5)
#Step 3- writer chain
print("\n" + "="*50)
print("step 3- Writer is drafting the report ...")
print("="*50)
research_combined= (
f"Search Results: \n {state['search_results']}\n\n"
f"Detailed Scraped Content: \n {state['scraped_content']}"
)
state['report']= writer_chain.invoke({
"topic":topic,
"research": research_combined
})
print("\n final report\n", state['report'])
#Critic Report
print("\n" + "="*50)
print("step 3- Critic is reviewing the report ...")
print("="*50)
# Introduce delay to prevent rate limits
time.sleep(5)
state['feedback']=critic_chain.invoke({
"report": state['report']
})
print("\n critic report \n", state['feedback'])
return state
if __name__ == "__main__":
topic= input("\n Enter a research topic: " )
run_research_pipeline(topic)
|