|
|
|
|
|
|
|
|
|
|
|
|
|
|
| from __future__ import annotations
|
|
|
| import argparse
|
| import os
|
| from typing import List, Optional, Dict, Any
|
|
|
|
|
| from langchain.agents import AgentExecutor, create_react_agent
|
| from langchain_core.prompts import PromptTemplate
|
| from langchain_core.runnables import RunnableConfig
|
| from langchain_community.utilities import RequestsWrapper
|
| from langchain.tools import Tool
|
|
|
|
|
| from langchain_huggingface import ChatHuggingFace, HuggingFaceEndpoint
|
|
|
|
|
| from langchain_community.tools import DuckDuckGoSearchRun, RequestsGetTool
|
|
|
|
|
| REACT_PROMPT = """You are a meticulous research assistant for GAIA-style fact questions.
|
|
|
| You have access to these tools:
|
| {tools}
|
|
|
| Use this format:
|
|
|
| Question: the input question to answer
|
| Thought: think about what to search or read next
|
| Action: one of [{tool_names}]
|
| Action Input: the input to the action
|
| Observation: the result of the action
|
| ... (repeat Thought/Action/Observation as needed)
|
| Thought: I can now answer.
|
| Final Answer: <the shortest exact answer>
|
|
|
| Follow the rules:
|
| - Prefer precise quotes from primary/official sources when possible.
|
| - Keep the final answer SHORT (a few words if possible)
|
| - If you need multiple pages, open them sequentially.
|
| - Never repeat the exact same Action Input twice. If an Action returns a login, auth, or error page, change strategy (use search with site: filters, try a different public page, or a cached/mirrored source).
|
| - If a URL contains "login", "@app/auth", or "returnto=", do NOT open it; search for a public page instead.
|
|
|
| Question: {input}
|
| {agent_scratchpad}"""
|
|
|
|
|
|
|
| class WebAgent:
|
| """Encapsulated GAIA-style web agent.
|
|
|
| Usage:
|
| web_agent = WebAgent()
|
| result = web_agent("Who is the CEO of ...?")
|
| # or
|
| result = web_agent.answer("...")
|
|
|
| You can also run as a script:
|
| python web_agent.py "Your question here"
|
| """
|
|
|
| def __init__(
|
| self,
|
| hf_repo_id: Optional[str] = None,
|
| max_new_tokens: int = 2048,
|
| do_sample: bool = False,
|
| repetition_penalty: float = 1.05,
|
| max_iterations: int = 20,
|
| verbose: bool = True,
|
| user_agent: str = "Mozilla/5.0 (compatible; GAIA/1.0; +https://example.org/bot)",
|
| ):
|
| self.hf_repo_id = hf_repo_id or os.environ.get("HF_REPO_ID", "Qwen/Qwen2.5-7B-Instruct")
|
| self.max_new_tokens = max_new_tokens
|
| self.do_sample = do_sample
|
| self.repetition_penalty = repetition_penalty
|
| self.max_iterations = max_iterations
|
| self.verbose = verbose
|
| self.user_agent = user_agent
|
|
|
|
|
| self._executor: Optional[AgentExecutor] = None
|
| self._tools: Optional[List[Tool]] = None
|
|
|
|
|
| def answer(self, question: str) -> str:
|
| """Run the agent and return the model's 'Final Answer' string."""
|
| executor = self._build_agent()
|
| result: Dict[str, Any] = executor.invoke({"input": question})
|
| return result["output"]
|
|
|
| def __call__(self, question: str) -> str:
|
| return self.answer(question)
|
|
|
|
|
| def _build_model(self) -> ChatHuggingFace:
|
| llm = HuggingFaceEndpoint(
|
| repo_id=self.hf_repo_id,
|
| task="text-generation",
|
| max_new_tokens=self.max_new_tokens,
|
| do_sample=self.do_sample,
|
| repetition_penalty=self.repetition_penalty,
|
| )
|
| return ChatHuggingFace(llm=llm)
|
|
|
| def _build_tools(self) -> List[Tool]:
|
| wrapper = RequestsWrapper(headers={"User-Agent": self.user_agent})
|
| raw_get = RequestsGetTool(
|
| name="Requests Get (raw)",
|
| requests_wrapper=wrapper,
|
| allow_dangerous_requests=True,
|
| )
|
|
|
| seen: set[str] = set()
|
|
|
| def guarded_get(url: str) -> str:
|
| low = url.lower()
|
| if any(x in low for x in ["@app/auth", "login?returnto=", "/login"]):
|
| return ("BLOCKED: This looks like a login/auth wall. "
|
| "Use search to find a public or cached page instead.")
|
| if url in seen:
|
| return (f"BLOCKED: Already fetched {url}. "
|
| "Choose a different URL or try a different query.")
|
| seen.add(url)
|
| return raw_get.run(url)
|
|
|
| return [
|
| DuckDuckGoSearchRun(name="DuckDuckGo Search"),
|
| Tool.from_function(
|
| name="Requests Get",
|
| func=guarded_get,
|
| description="HTTP GET for public pages; prevents repeats and login walls.",
|
| ),
|
| ]
|
|
|
| def _build_agent(self) -> AgentExecutor:
|
| model = self._build_model()
|
| tools = self._build_tools()
|
| prompt = PromptTemplate.from_template(REACT_PROMPT)
|
|
|
| agent = create_react_agent(model, tools, prompt)
|
| executor = AgentExecutor(
|
| agent=agent,
|
| tools=tools,
|
| max_iterations=self.max_iterations,
|
| handle_parsing_errors=True,
|
| verbose=self.verbose,
|
| )
|
| return executor
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| def main():
|
| parser = argparse.ArgumentParser(description="Class-based LangChain web agent for GAIA-style questions.")
|
| parser.add_argument("question", type=str, help="Your question (quoted).")
|
| args = parser.parse_args()
|
|
|
| web_agent = WebAgent()
|
| print(web_agent.answer(args.question))
|
|
|
|
|
| if __name__ == "__main__":
|
| main()
|
|
|
