commit all files

z_http_request.py

@@ -0,0 +1,31 @@
+import html2text
+import requests
+
+def get_http_request(url: str, params: dict = None) -> str:
+    """ use this http request tool visiting website with http url to get more information.
+    for a given url(parameter), send a http request
+    to get more information in the url
+    """
+    max_return_char = 28000
+    try:
+        headers = {
+            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36"
+        }
+        response = requests.get(url, params=params,headers=headers)
+        response.raise_for_status()
+        converter = html2text.HTML2Text()
+        converter.ignore_images = True
+        converter.ignore_links = True
+        # converter.ignore_tables = True
+        markdown_text = converter.handle(response.text)
+        if len(markdown_text) > max_return_char:
+            print(f"\n http result size is : {len(markdown_text)}. try to truncate it\n")
+            markdown_text = markdown_text[:max_return_char] + "...[truncated]"
+        return markdown_text
+    except requests.exceptions.RequestException as e:
+        return f"http get_request error: {e}"
+
+
+if __name__ == "__main__" :
+    html = get_http_request(url = "https://www.orczhou.com")
+    print(html)

z_search.py

@@ -0,0 +1,63 @@
+import os
+from googleapiclient.discovery import build
+from ddgs import DDGS
+
+GOOGLE_CSE_ID = os.getenv("GOOGLE_CSE_ID")
+GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY")
+
+def tool_google_search(query: str):
+    """
+    This tool is a web search utility that leverages the DuckDuckGo search engine. 
+    Its primary purpose is to allow a large language model (LLM) to retrieve real-time, up-to-date information from the internet.
+
+    Args:
+        query (str): search string
+
+    Returns:
+        string: top 3 related searching results
+    """
+    num_results = 3
+    try:
+        service = build("customsearch", "v1", developerKey=GOOGLE_API_KEY)
+
+        res = service.cse().list(
+            q=query,
+            cx=GOOGLE_CSE_ID,
+            num=num_results
+        ).execute()
+
+        r = ""
+        if 'items' in res:
+            search_results = res['items']
+            for i,item in enumerate(search_results,1):
+                r += f"\n search result title {i}:" + item.get('title') + "\n"
+                r += f"\n search result summary {i}: " + item.get('snippet') + "\n"
+                r += f"\n get more information with this url: " +  item.get('link') + " \n\n"
+            return r
+        else:
+            r = "no search result."
+            return r
+            
+    except Exception as e:
+        print(f"something is wrong: {e}")
+        return ""
+
+def search_information_from_web(query: str) -> str:
+    """ This tool is a web search utility that leverages the DuckDuckGo search engine. Its primary purpose is to allow a large language model (LLM) to retrieve real-time, up-to-date information from the internet.
+    input variables:
+         query  string
+    """
+    results = DDGS().text(query, max_results=3)
+    r = ""
+    if results:
+        for i,result in enumerate(results, 1):
+            r += f"\n search result summary {i}: {result['body']}"
+            r += f"\n get more information with this url: {result['href']} \n\n"
+    else:
+        r = "no search result."
+    return r
+
+if __name__ == "__main__":
+    search_query = "who is orczhou"
+    results = tool_google_search(search_query)
+    print(results)

zex.py

@@ -0,0 +1,108 @@
+# version 0.3
+
+from typing_extensions import TypedDict
+from typing import List, TypedDict, Annotated, Optional
+from langgraph.prebuilt import ToolNode, tools_condition
+from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage
+from langchain_openai import ChatOpenAI
+from langgraph.graph.message import add_messages
+from langchain_core.messages import SystemMessage, HumanMessage
+from langgraph.graph import START, StateGraph
+import os
+
+#############   state    #############
+class AgentState(TypedDict):
+    task_id : str
+    question: str
+    file_name: str
+    file_url: str
+    messages: Annotated[list[AnyMessage], add_messages]
+
+builder = StateGraph(AgentState)
+
+#######  tools and tools node  #######
+
+from z_http_request import get_http_request
+from z_search import tool_google_search,search_information_from_web
+
+tools = [tool_google_search,get_http_request]
+# tools = [search_information_from_web,get_http_request]
+
+builder.add_node("tools", ToolNode(tools))
+
+############# LLM instance  ##########
+# model = "gpt-o3"
+# model = "gpt-4o"
+model = "gpt-4.1"
+OPENAI_KEY=os.getenv("OPENAI_KEY")
+llm = ChatOpenAI(model=model,api_key=OPENAI_KEY)
+llm_with_tools = llm.bind_tools(tools, parallel_tool_calls=False)
+
+############# the only node ##########
+# node analysis
+
+def analysis(state: AgentState):
+    if not state["messages"]:
+        # first time call analysis
+        # no extra file at all
+        user_message = HumanMessage(content=f"{state['question']}")
+        state["messages"].append(user_message)
+        print(state)
+
+    sys_prompt = """You solve GAIA Level-1 tasks.
+Return ONLY the final answer string (no extra words).Match requested format exactly.
+Here are some example:
+  question: 3+8=? ;   answer is : 11 (note: answer is not "3+8=11")
+  question: who is the president of USA? ;  answer is : Donult Trump (note: answer is not "the president of USA is Donult Trump")
+  if the question is like "how many ...", just answer the number ONLY
+  if the question is like "where ...", just answer the location ONLY
+Use web search tool only when needed. 
+Use http request tool if nessesary to get more detailed information after the searching
+Use extract_text tool to get text from a image if there is a image as an attachment
+if the question is writter in other language, directly answer it with the language.
+when writing a comma-delimited list, you should format it as item1, item2, item3—not item1,item2,item3
+
+You do not know what date is today. So, if there is any question about date,try to search result for it.
+
+Strip leading/trailing whitespace from the final answer.
+"""
+    sys_msg = SystemMessage(content=sys_prompt)
+
+    import time
+    from openai import RateLimitError
+
+    max_retries = 3
+    retry_delay = 3
+    
+    for attempt in range(max_retries):
+        try:
+            response = llm_with_tools.invoke([sys_msg] + state["messages"])
+            break
+        except RateLimitError as e:
+            print(f"try {attempt + 1} failed：{e}")
+            if attempt < max_retries - 1:
+                print(f"wait {retry_delay} seconds ...")
+                time.sleep(retry_delay)
+                print(f"let's try again ...")
+                retry_delay *= 2 
+            else:
+                print("max retry times")
+                raise  
+
+    # response = llm_with_tools.invoke([sys_msg] + state["messages"])
+    return {"messages": [response]}
+
+builder.add_node("analysis", analysis)
+
+############  edges   ################
+builder.add_edge(START, "analysis")
+builder.add_conditional_edges(
+    "analysis",
+    # If the latest message requires a tool, route to tools
+    # Otherwise, provide a direct response
+    tools_condition,
+)
+builder.add_edge("tools", "analysis")
+
+######### compile the agent ##########
+zex_vs_gaia_graph = builder.compile()