DeepResearch代码浅析

DeepResearch代码浅析

概述

代码：DeepResearch

主要看一下inference下面的ReAct推理流程。

1. inference
2. ├── eval_data
3. │ ├── example_with_file.jsonl
4. │ ├── example.jsonl
5. │ └── file_corpus
6. │ └── hello.txt
7. ├── file_tools
8. │ ├── __pycache__
9. │ │ └── file_parser.cpython-313.pyc
10. │ ├── file_parser.py
11. │ ├── idp.py
12. │ ├── utils.py
13. │ ├── video_agent.py
14. │ └── video_analysis.py
15. ├── prompt.py
16. ├── react_agent.py
17. ├── run_multi_react.py
18. ├── run_react_infer.sh
19. ├── tool_file.py
20. ├── tool_python.py
21. ├── tool_scholar.py
22. ├── tool_search.py
23. └── tool_visit.py

复制代码

代码的入口是run_react_infer.sh中的run_multi_react.py文件

run_multi_react.py负责初始化节点环境，加载数据集，加载模型配置，进行多次rollout采样。

react_agent是ReAct 架构的Agent，负责迭代输出，调用工具。

1. from react_agent import MultiTurnReactAgent
3. test_agent = MultiTurnReactAgent(
4. llm=llm_cfg,
5. function_list=["search", "visit", "google_scholar", "PythonInterpreter"]
6. )

复制代码

react_agent

主体的ReAct agent，统一调度处理模型的输出，进行tool extract and execute和tool response的拼接

执行ReAct的全部流程，给出最后的执行状态，处理运行中的异常现象

• 定义工具

  1. from tool_file import *
  2. from tool_scholar import *
  3. from tool_python import *
  4. from tool_search import *
  5. from tool_visit import *
  7. OBS_START = '<tool_response>'
  8. OBS_END = '\n</tool_response>'
  10. # 定义工具，放在TOOL_MAP中
  11. TOOL_CLASS = [
  12. FileParser(),
  13. Scholar(),
  14. Visit(),
  15. Search(),
  16. PythonInterpreter(),
  17. ]
  18. TOOL_MAP = {tool.name: tool for tool in TOOL_CLASS}

  复制代码

• 在MultiTurnReactAgent类中使用def call_server() 调用llm api

  1. def call_server(self, msgs, planning_port, max_tries=10):
  3. openai_api_key = "EMPTY"
  4. openai_api_base = f"http://127.0.0.1:{planning_port}/v1"
  6. client = OpenAI(
  7. api_key=openai_api_key,
  8. base_url=openai_api_base,
  9. timeout=600.0,
  10. )

  复制代码

• 执行ReAct流程

  可能出现的情况

  • 返回answer （出现 ）
    • 未达到轮次限制
    • 达到/未达到上下文token数量限制
  • 未返回answer
    • 超出轮次限制后
    • 达到上下问token数量限制后，返回答案没有
    • 超时
    • 工具调用错误：tool_call 的json格式错误
  1. def _run(self, data: str, model: str, **kwargs) -> List[List[Message]]:
  2. #############################################################
  3. # 初始化question和最多调用轮次num_llm_calls_available，
  4. # 记录start_time，拼接最开始的message
  5. #############################################################
  6. self.model=model
  7. try:
  8. question = data['item']['question']
  9. except:
  10. raw_msg = data['item']['messages'][1]["content"]
  11. question = raw_msg.split("User:")[1].strip() if "User:" in raw_msg else raw_msg
  13. start_time = time.time()
  14. planning_port = data['planning_port']
  15. answer = data['item']['answer']
  16. self.user_prompt = question
  17. system_prompt = SYSTEM_PROMPT
  18. cur_date = today_date()
  19. system_prompt = system_prompt + str(cur_date)
  20. messages = [{"role": "system", "content": system_prompt}, {"role": "user", "content": question}]
  21. num_llm_calls_available = MAX_LLM_CALL_PER_RUN
  22. round = 0
  23. #############################################################
  24. # 开始迭代每一个iter，生成<tool_call> 或是
  25. #############################################################
  26. while num_llm_calls_available > 0:
  27. # Check whether time is reached
  28. #############################################################
  29. # 检查是否超时（2.5小时）
  30. #############################################################
  31. if time.time() - start_time > 150 * 60: # 150 minutes in seconds
  32. prediction = 'No answer found after 2h30mins'
  33. termination = 'No answer found after 2h30mins'
  34. result = {
  35. "question": question,
  36. "answer": answer,
  37. "messages": messages,
  38. "prediction": prediction,
  39. "termination": termination
  40. }
  41. return result
  42. #############################################################
  43. # 更新调用llm次数 num_llm_calls_available
  44. # 获取llm的返回值 content
  45. #############################################################
  46. round += 1
  47. num_llm_calls_available -= 1
  48. content = self.call_server(messages, planning_port)
  49. print(f'Round {round}: {content}')
  50. #############################################################
  51. # 进行content中关键tool的提取
  52. #############################################################
  53. # 舍弃content中<tool_response>的部分，应为obs应该是user输入的，而不是llm生成的
  54. if '<tool_response>' in content:
  55. pos = content.find('<tool_response>')
  56. content = content[:pos]
  57. messages.append({"role": "assistant", "content": content.strip()})
  58. # 查看content中是否有工具调用 <tool_call>
  59. if '<tool_call>' in content and '</tool_call>' in content:
  60. tool_call = content.split('<tool_call>')[1].split('</tool_call>')[0]
  61. try:
  62. # 使用python解释器运行code_raw
  63. if "python" in tool_call.lower():
  64. try:
  65. code_raw=content.split('<tool_call>')[1].split('</tool_call>')[0].split('')[1].split('')[0].strip()
  66. result = TOOL_MAP['PythonInterpreter'].call(code_raw)
  67. except:
  68. result = "[Python Interpreter Error]: Formatting error."
  69. # 调用其他的工具
  70. else:
  71. tool_call = json5.loads(tool_call)
  72. tool_name = tool_call.get('name', '')
  73. tool_args = tool_call.get('arguments', {})
  74. result = self.custom_call_tool(tool_name, tool_args)
  76. # 如果llm生成的tool formart错误，则将错误信息写入messages中（可以使用约束采样避免格式错误）
  77. except:
  78. result = 'Error: Tool call is not a valid JSON. Tool call must contain a valid "name" and "arguments" field.'
  79. result = "<tool_response>\n" + result + "\n</tool_response>"
  80. # print(result)
  81. # 把tool response写入到user中
  82. messages.append({"role": "user", "content": result})
  83. # 如果模型生成的content中有 </answer>，则已经输出答案
  84. if '' in content and '</answer>' in content:
  85. termination = 'answer'
  86. break
  87. # 如果没有可用轮次，记录失败信息
  88. if num_llm_calls_available <= 0 and '' not in content:
  89. messages[-1]['content'] = 'Sorry, the number of llm calls exceeds the limit.'
  91. max_tokens = 110 * 1024
  92. token_count = self.count_tokens(messages)
  93. print(f"round: {round}, token count: {token_count}")
  95. #############################################################
  96. # ReAct的累积上下文token长度达到阈值，强制给出回答
  97. #############################################################
  98. if token_count > max_tokens:
  99. print(f"Token quantity exceeds the limit: {token_count} > {max_tokens}")
  101. messages[-1]['content'] = "You have now reached the maximum context length you can handle. You should stop making tool calls and, based on all the information above, think again and provide what you consider the most likely answer in the following format:<think>your final thinking</think>\nyour answer</answer>"
  102. content = self.call_server(messages, planning_port)
  103. messages.append({"role": "assistant", "content": content.strip()})
  104. # token数达到阈值后，成功返回结果
  105. if '' in content and '</answer>' in content:
  106. prediction = messages[-1]['content'].split('')[1].split('</answer>')[0]
  107. termination = 'generate an answer as token limit reached'
  108. # 未返回结果
  109. else:
  110. prediction = messages[-1]['content']
  111. termination = 'format error: generate an answer as token limit reached'
  112. result = {
  113. "question": question,
  114. "answer": answer,
  115. "messages": messages,
  116. "prediction": prediction,
  117. "termination": termination
  118. }
  119. return result
  121. # 这里termination忽略了token超限制后是否给出answer的情况
  122. if '' in messages[-1]['content']:
  123. prediction = messages[-1]['content'].split('')[1].split('</answer>')[0]
  124. termination = 'answer'
  125. else:
  126. prediction = 'No answer found.'
  127. termination = 'answer not found'
  128. if num_llm_calls_available == 0:
  129. termination = 'exceed available llm calls'
  130. result = {
  131. "question": question,
  132. "answer": answer,
  133. "messages": messages,
  134. "prediction": prediction,
  135. "termination": termination
  136. }
  137. return result

  复制代码

工具调用

• tool_python

  执行python代码。\((code;Interpreter)\rightarrow (stdout, stderr)\)

  1. def call(self, params, files= None, timeout = 50, **kwargs) -> str:
  2. try:
  3. # params 即为要执行的code代码
  4. code=params
  5. last_error = None
  6. # 尝试多次
  7. for attempt in range(8):
  8. try:
  9. # Randomly sample an endpoint for each attempt
  10. endpoint = random.choice(SANDBOX_FUSION_ENDPOINTS)
  11. print(f"Attempt {attempt + 1}/5 using endpoint: {endpoint}")
  13. # 执行code
  14. code_result = run_code(RunCodeRequest(code=code, language='python', run_timeout=timeout), max_attempts=1, client_timeout=timeout, endpoint=endpoint)
  15. print("[Python] Code Result", code_result)
  16. result = []
  17. # 记录code 的标准输出和错误
  18. if code_result.run_result.stdout:
  19. result.append(f"stdout:\n{code_result.run_result.stdout}")
  20. if code_result.run_result.stderr:
  21. result.append(f"stderr:\n{code_result.run_result.stderr}")
  22. if code_result.run_result.execution_time >= timeout-1:
  23. result.append(f"[PythonInterpreter Error] TimeoutError: Execution timed out.")
  24. result = '\n'.join(result)
  25. print('SUCCESS RUNNING TOOL')
  26. return result if result.strip() else 'Finished execution.'
  28. # code执行超时
  29. except Timeout as e:
  30. last_error = f'[Python Interpreter Error] TimeoutError: Execution timed out on endpoint {endpoint}.'
  31. print(f"Timeout on attempt {attempt + 1}: {last_error}")
  32. if attempt == 4: # Last attempt
  33. return last_error
  34. continue
  36. # code执行错误
  37. except Exception as e:
  38. last_error = f'[Python Interpreter Error]: {str(e)} on endpoint {endpoint}'
  39. print(f"Error on attempt {attempt + 1}: {last_error}")
  40. if attempt == 4: # Last attempt
  41. return last_error
  42. continue
  44. return last_error if last_error else '[Python Interpreter Error]: All attempts failed.'
  46. except Exception as e:
  47. return f"[Python Interpreter Error]: {str(e)}"

  复制代码

• tool_visit

搜索具体的url，并根据goal总结返回。\((url, goal;\pi)\rightarrow summary\)

1. JINA_API_KEYS = os.getenv("JINA_API_KEYS", "")
4. def readpage_jina(self, url: str, goal: str) -> str:
5. """
6. Attempt to read webpage content by alternating between jina and aidata services.
8. Args:
9. url: The URL to read
10. goal: The goal/purpose of reading the page
12. Returns:
13. str: The webpage content or error message
14. """
16. # def call_server用于根据goal总结网页的内容
17. summary_page_func = self.call_server
18. max_retries = int(os.getenv('VISIT_SERVER_MAX_RETRIES', 1))
20. # 使用jina将url的网页信息转化为 markdown格式
21. content = self.html_readpage_jina(url)
23. #############################################################
24. # 处理markdown的网页信息 content
25. #############################################################
26. # 如果网页信息可以被jina提取
27. if content and not content.startswith("[visit] Failed to read page.") and content != "[visit] Empty content." and not content.startswith("[document_parser]"):
28. # pre-process 先处理content的token长度，避免llm的上下文超长
29. content = truncate_to_tokens(content, max_tokens=95000)
30. # 总结promopt
31. messages = [{"role":"user","content": EXTRACTOR_PROMPT.format(webpage_content=content, goal=goal)}]
32. parse_retry_times = 0
33. # 得到网页总结后的信息 raw
34. raw = summary_page_func(messages, max_retries=max_retries)
35. summary_retries = 3
36. # 如果raw少于10个字符，那么认为总结失败，因为raw是json格式，```json {"rational":..., "evidence":..., "summary":...}```
37. while len(raw) < 10 and summary_retries >= 0:
38. # 尝试截断30%的长度
39. truncate_length = int(0.7 * len(content)) if summary_retries > 0 else 25000
40. status_msg = (
41. f"[visit] Summary url[{url}] "
42. f"attempt {3 - summary_retries + 1}/3, "
43. f"content length: {len(content)}, "
44. f"truncating to {truncate_length} chars"
45. ) if summary_retries > 0 else (
46. f"[visit] Summary url[{url}] failed after 3 attempts, "
47. f"final truncation to 25000 chars"
48. ) # 截断30%不行，尝试只留下25000字符
49. print(status_msg)
50. content = content[:truncate_length]
51. extraction_prompt = EXTRACTOR_PROMPT.format(
52. webpage_content=content,
53. goal=goal
54. )
55. messages = [{"role": "user", "content": extraction_prompt}]
56. raw = summary_page_func(messages, max_retries=max_retries)
57. summary_retries -= 1
59. # 解析总结的格式
60. parse_retry_times = 2
61. if isinstance(raw, str):
62. raw = raw.replace("```json", "").replace("```", "").strip()
63. while parse_retry_times < 3:
64. try:
65. raw = json.loads(raw)
66. break
67. except:
68. # 解析失败的话，就重新生成总结
69. raw = summary_page_func(messages, max_retries=max_retries)
70. parse_retry_times += 1
72. # 解析失败
73. if parse_retry_times >= 3:
74. useful_information = "The useful information in {url} for user goal {goal} as follows: \n\n".format(url=url, goal=goal)
75. useful_information += "Evidence in page: \n" + "The provided webpage content could not be accessed. Please check the URL or file format." + "\n\n"
76. useful_information += "Summary: \n" + "The webpage content could not be processed, and therefore, no information is available." + "\n\n"
77. # 解析成功，把evidence和summary一并返回
78. else:
79. useful_information = "The useful information in {url} for user goal {goal} as follows: \n\n".format(url=url, goal=goal)
80. useful_information += "Evidence in page: \n" + str(raw["evidence"]) + "\n\n"
81. useful_information += "Summary: \n" + str(raw["summary"]) + "\n\n"
83. if len(useful_information) < 10 and summary_retries < 0:
84. print("[visit] Could not generate valid summary after maximum retries")
85. useful_information = "[visit] Failed to read page"
87. return useful_information
89. # If no valid content was obtained after all retries
90. # 如果网页的原始信息就不合理，jina无法提取，返回失败信息
91. else:
92. useful_information = "The useful information in {url} for user goal {goal} as follows: \n\n".format(url=url, goal=goal)
93. useful_information += "Evidence in page: \n" + "The provided webpage content could not be accessed. Please check the URL or file format." + "\n\n"
94. useful_information += "Summary: \n" + "The webpage content could not be processed, and therefore, no information is available." + "\n\n"
95. return useful_information

复制代码

jina举例

输入https://r.jina.ai/+{url(https://www.axtonliu.ai/newsletters/ai-2/posts/jina-reader-api-four-usage-methods-guide)}

原始网页：

jina由三部分组成：

• title
• url
• markdown content（图片的url信息，超链接等）

1. Title: Jina Reader API完全指南：4种实用集成方案详解 | AI开发教程
3. URL Source: https://www.axtonliu.ai/newsletters/ai-2/posts/jina-reader-api-four-usage-methods-guide
5. Markdown Content:
6. 构建知识库，或者分析各种文章数据，是大家使用 AI 很重要的一个应用场景，

复制代码

• tool_file

  根据url的文件，和goal，返回总结信息，类似于tool_visit。但是要借助于file_tools进行指定url文件的读取（visit是借用jina进行指定url网页信息的读取）。

  1. """
  2. input:
  3. - query/goal: str
  4. - Docs: List[file]/List[url]
  5. - file type: 'pdf', 'docx', 'pptx', 'txt', 'html', 'csv', 'tsv', 'xlsx', 'xls', 'doc', 'zip', '.mp4', '.mov', '.avi', '.mkv', '.webm', '.mp3', '.wav', '.aac', '.ogg', '.flac'
  6. output:
  7. - answer: str
  8. - useful_information: str
  9. """

  复制代码

• tool_search

  调用google 进行search。\((q;Enginer)\rightarrow docs\)

• tool_scholar

  类似于tool_search，区别在于 tool_scholar在goole scholar上进行文章的搜索

Prompt

分为react的system prompt，以及visit 总结的extract prompt

1. SYSTEM_PROMPT = """You are a deep research assistant. Your core function is to conduct thorough, multi-source investigations into any topic. You must handle both broad, open-domain inquiries and queries within specialized academic fields. For every request, synthesize information from credible, diverse sources to deliver a comprehensive, accurate, and objective response. When you have gathered sufficient information and are ready to provide the definitive response, you must enclose the entire final answer within </answer> tags.
3. # Tools
5. You may call one or more functions to assist with the user query.
7. You are provided with function signatures within <tools></tools> XML tags:
8. <tools>
9. {"type": "function", "function": {"name": "search", "description": "Perform Google web searches then returns a string of the top search results. Accepts multiple queries.", "parameters": {"type": "object", "properties": {"query": {"type": "array", "items": {"type": "string", "description": "The search query."}, "minItems": 1, "description": "The list of search queries."}}, "required": ["query"]}}}
10. {"type": "function", "function": {"name": "visit", "description": "Visit webpage(s) and return the summary of the content.", "parameters": {"type": "object", "properties": {"url": {"type": "array", "items": {"type": "string"}, "description": "The URL(s) of the webpage(s) to visit. Can be a single URL or an array of URLs."}, "goal": {"type": "string", "description": "The specific information goal for visiting webpage(s)."}}, "required": ["url", "goal"]}}}
11. {"type": "function", "function": {"name": "PythonInterpreter", "description": "Executes Python code in a sandboxed environment. To use this tool, you must follow this format:
12. 1. The 'arguments' JSON object must be empty: {}.
13. 2. The Python code to be executed must be placed immediately after the JSON block, enclosed within and tags.
15. IMPORTANT: Any output you want to see MUST be printed to standard output using the print() function.
17. Example of a correct call:
18. <tool_call>
19. {"name": "PythonInterpreter", "arguments": {}}
21. import numpy as np
22. # Your code here
23. print(f"The result is: {np.mean([1,2,3])}")
25. </tool_call>", "parameters": {"type": "object", "properties": {}, "required": []}}}
26. {"type": "function", "function": {"name": "google_scholar", "description": "Leverage Google Scholar to retrieve relevant information from academic publications. Accepts multiple queries. This tool will also return results from google search", "parameters": {"type": "object", "properties": {"query": {"type": "array", "items": {"type": "string", "description": "The search query."}, "minItems": 1, "description": "The list of search queries for Google Scholar."}}, "required": ["query"]}}}
27. {"type": "function", "function": {"name": "parse_file", "description": "This is a tool that can be used to parse multiple user uploaded local files such as PDF, DOCX, PPTX, TXT, CSV, XLSX, DOC, ZIP, MP4, MP3.", "parameters": {"type": "object", "properties": {"files": {"type": "array", "items": {"type": "string"}, "description": "The file name of the user uploaded local files to be parsed."}}, "required": ["files"]}}}
28. </tools>
30. For each function call, return a json object with function name and arguments within <tool_call></tool_call> XML tags:
31. <tool_call>
32. {"name": <function-name>, "arguments": }
33. </tool_call>
35. Current date: """
37. EXTRACTOR_PROMPT = """Please process the following webpage content and user goal to extract relevant information:
39. ## **Webpage Content**
40. {webpage_content}
42. ## **User Goal**
43. {goal}
45. ## **Task Guidelines**
46. 1. **Content Scanning for Rational**: Locate the **specific sections/data** directly related to the user's goal within the webpage content
47. 2. **Key Extraction for Evidence**: Identify and extract the **most relevant information** from the content, you never miss any important information, output the **full original context** of the content as far as possible, it can be more than three paragraphs.
48. 3. **Summary Output for Summary**: Organize into a concise paragraph with logical flow, prioritizing clarity and judge the contribution of the information to the goal.
50. **Final Output Format using JSON format has "rational", "evidence", "summary" feilds**
51. """

复制代码

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！