| """ |
| 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 |
|
|
