Spaces:
Running
Running
| from __future__ import annotations | |
| from src.schemas import ExternalAgentResult, SanitizedTask | |
| from src.tools.fetch import fetch_text | |
| from src.tools.search import search_web | |
| class DirectSearchAgent: | |
| def run(self, task: SanitizedTask) -> ExternalAgentResult: | |
| sources = search_web(task.sanitized_query) | |
| errors: list[str] = [] | |
| if not sources: | |
| errors.append("Live direct search returned no sources. No fixture fallback was used because MOCK_MODE=false.") | |
| for source in sources: | |
| if source.extracted_text is None and source.url: | |
| source.extracted_text, source.fetch_status = fetch_text(source.url) | |
| if sources: | |
| summary = _summarize_sources(sources) | |
| else: | |
| summary = "I could not find a reliable live result for that." | |
| return ExternalAgentResult(route="direct_search", summary=summary[:1200], sources=sources, errors=errors) | |
| def _summarize_sources(sources) -> str: | |
| first = sources[0] | |
| title = first.title.strip() or "the top result" | |
| snippet = (first.snippet or first.extracted_text or "").strip() | |
| snippet = snippet.replace("...", "").strip() | |
| if len(snippet) > 260: | |
| snippet = snippet[:257].rsplit(" ", 1)[0] + "." | |
| source_names = [source.title.strip() for source in sources[:3] if source.title.strip()] | |
| if len(source_names) > 1: | |
| names = ", ".join(source_names[:-1]) + f", and {source_names[-1]}" | |
| elif source_names: | |
| names = source_names[0] | |
| else: | |
| names = "the live results" | |
| if snippet: | |
| return f"The clearest result I found is {title}. {snippet} I also found {names}." | |
| return f"The clearest result I found is {title}. I also found {names}." | |