File size: 2,759 Bytes
40deb66 cee18af bcdc55d 40deb66 a394be7 40deb66 bcdc55d 40deb66 cee18af 40deb66 cee18af a394be7 cee18af a394be7 40deb66 | 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 | from datetime import datetime, timezone
from colorama import Fore, Style # type: ignore[import]
from langchain_core.tools import tool
from langchain.agents import create_agent
from langgraph.errors import GraphRecursionError
from agent.api.api import get_llm
from agent.tools.search import web_search
@tool
def websearch_agent(query: str) -> str:
"""
A single web search agent that searches the internet and returns an answer.
Use this tool when you need to find real-time or factual information from the web.
Pros:
- Has continuous memory across search steps, allowing deep investigation on a single topic.
Cons:
- Narrow field of view, can only follow one search thread at a time.
- May fail after too many steps due to token limit overflow.
Prefer websearch_agents for complex questions requiring broad, multi-source research.
Use this tool for simple, direct factual lookups.
Args:
query: The question or search query to look up on the web.
"""
print(f"{Fore.YELLOW}[SupervisorAgent -> WebSearchAgent] {query}{Style.RESET_ALL}")
base_agent = create_agent(
model=get_llm(),
tools=[web_search],
system_prompt=(
f"Current time is: {datetime.now(timezone.utc).isoformat()}. "
f"Your memory are out of date. "
f"All of truth that you believe without search are wrong. "
f"You must search the web and find the lastest answer."
f"Just run 1 turn search. "
),
)
try:
result = base_agent.invoke(
{"messages": [{"role": "user", "content": query}]},
# config={"recursion_limit": 10},
)
content = result["messages"][-1].content
if isinstance(content, list):
content = content[0].get("text", "")
else:
content = str(content)
except GraphRecursionError:
print(
f"{Fore.RED}[WebSearchAgent] Recursion limit reached, returning partial results.{Style.RESET_ALL}"
)
content = "Search completed but no definitive answer was found within the allowed steps."
except Exception as e:
error_msg = str(e)
print(f"{Fore.RED}[WebSearchAgent] Error: {error_msg}{Style.RESET_ALL}")
content = (
f"Search agent failed with error: {error_msg}. "
f"Recommend retrying with the web_search_agents tool to avoid context length overflow."
)
print(
f"{Fore.YELLOW}[WebSearchAgent -> SupervisorAgent] {content}{Style.RESET_ALL}"
)
return content
if __name__ == "__main__":
from dotenv import load_dotenv
load_dotenv()
answer = websearch_agent.invoke({"query": "What is LangGraph?"})
print(answer)
|