File size: 3,119 Bytes
939c0c0 | 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 | """
Web Search Specialist Agent Node
=================================
Performs free external web search using DuckDuckGo search integration.
Summarizes search results with clickable markdown URLs.
"""
from __future__ import annotations
import structlog
from agents.state import CopilotState
logger = structlog.get_logger(__name__)
def duckduckgo_search(query: str, max_results: int = 5) -> list[dict]:
"""Execute free DuckDuckGo search using duckduckgo_search library."""
try:
from duckduckgo_search import DDGS
with DDGS() as ddgs:
results = list(ddgs.text(query, max_results=max_results))
return [
{
"title": r.get("title", "Web Result"),
"link": r.get("href", ""),
"snippet": r.get("body", "")
}
for r in results
]
except Exception as e:
logger.warning("DuckDuckGo search failed or rate-limited", error=str(e))
return []
async def web_node(state: CopilotState) -> CopilotState:
"""
Web Search Specialist Node.
Searches the public web for real-time information.
"""
query = state.get("query", "")
logger.info("Web Agent executing search", query=query)
try:
results = duckduckgo_search(query, max_results=4)
chunks = []
citations = []
if results:
context_text = "### Web Search Results:\n"
for r in results:
context_text += f"- [{r['title']}]({r['link']}): {r['snippet']}\n"
citations.append({
"document_id": "web_search",
"document_name": r["title"],
"chunk_text": r["snippet"],
"score": 0.9,
"doc_type": "web"
})
chunks.append({
"document_id": "web_search_summary",
"document_name": "DuckDuckGo Search",
"text": context_text,
"score": 1.0,
"doc_type": "web"
})
else:
chunks.append({
"document_id": "web_search_none",
"document_name": "DuckDuckGo Search",
"text": "Web search returned no external results for this query.",
"score": 0.0,
"doc_type": "web"
})
state["retrieved_chunks"] = chunks
state["citations"] = citations
state["active_agent"] = "web"
outputs = state.get("agent_outputs", [])
outputs.append({
"agent_name": "web",
"content": f"Performed web search. Found {len(results)} results.",
"sources": citations
})
state["agent_outputs"] = outputs
except Exception as e:
logger.error("Web Agent error", error=str(e))
state["error"] = f"Web Search Error: {str(e)}"
state["retrieved_chunks"] = [{
"document_name": "Web Search Error",
"text": f"Web search encountered an issue: {str(e)}"
}]
return state
|