Class WebWalker

WebAgent/WebWalker/src/agent.py:19–208

This explorer agent use ReAct format to call tools

    'What is the {name_for_human} API useful for? {description_for_model} Parameters: {parameters} {args_format}')

class WebWalker(FnCallAgent):
    """This explorer agent use ReAct format to call tools"""

    def __init__(self,
                 function_list: Optional[List[Union[str, Dict, BaseTool]]] = None,
                 llm: Optional[Union[Dict, BaseChatModel]] = None,
                 system_message: Optional[str] = DEFAULT_SYSTEM_MESSAGE,
                 name: Optional[str] = None,
                 description: Optional[str] = None,
                 files: Optional[List[str]] = None,
                 **kwargs):
        super().__init__(function_list=function_list,
                         llm=llm,
                         system_message=system_message,
                         name=name,
                         description=description,
                         files=files,
                         **kwargs)
        self.extra_generate_cfg = merge_generate_cfgs(
            base_generate_cfg=self.extra_generate_cfg,
            new_generate_cfg={'stop': ['Observation:', 'Observation:\n']},
        )
        self.client = OpenAI(
            api_key=llm['api_key'], 
            base_url=llm['model_server'],
        )
        self.llm_cfg = llm
        self.momery = []

    def observation_information_extraction(self, query, observation):
        user_prompt = "- Query: {query}\n- Observation: {observation}".format(query=query, observation=observation)
        messages = [
            {'role': 'system', 'content': STSTEM_CRITIIC_INFORMATION},
            {'role': 'user', 'content':  user_prompt}]
        max_retries = 10
        for attempt in range(max_retries):
            try:
                response = self.client.chat.completions.create(
                    model=self.llm_cfg['model'],
                    response_format={"type": "json_object"},
                    messages=messages
                )
                print(response.choices[0].message.content)
                # response_content = json.loads(response.choices[0].message.content)
                if "true" in response.choices[0].message.content:
                    try:
                        return json.loads(response.choices[0].message.content)["information"]
                    except:
                        return response.choices[0].message.content
                else:
                    return None
            except Exception as e:
                print(e)
                if attempt < max_retries - 1:
                    time.sleep(1 * (2 ** attempt))  # Exponential backoff
                else:
                    raise e  # Raise the exception if the last retry fails