Langchain-Chatchat project: 5.1-ChatGLM3-6B tool call

Evaluations on data sets from different perspectives such as semantics, mathematics, reasoning, code, and knowledge show that ChatGLM3-6B-Base has the strongest performance among basic models below 10B. ChatGLM3-6B adopts a newly designed Prompt format, in addition to normal multi-round conversations. At the same time, it natively supports complex scenarios such as tool calling (Function Call), code execution (Code Interpreter), and Agent tasks. This article mainly introduces the registration of new tools in tool_registry.py to enhance model capabilities through weather query examples.

You can directly call LangChain’s own tools (for example, ArXiv), or you can call custom tools. Some of the tools that come with LangChain[2] are as follows:

1. Customized weather query tool
1.Weather class
You can refer to the Tool/Weather.py and Tool/Weather.yaml files, inherit the BaseTool class, and overload the _run() method, as shown below:

class Weather(BaseTool): # Weather query tool
    name = "weather"
    description = "Use for searching weather at a specific location"

    def __init__(self):
        super().__init__()

    def get_weather(self, location):
        api_key = os.environ["SENIVERSE_KEY"]
        url = f"https://api.seniverse.com/v3/weather/now.json?key={api_key} & amp;location={location} & amp;language=zh-Hans & amp;unit=c "
        response = requests.get(url)
        if response.status_code == 200:
            data = response.json()
            weather = {
                "temperature": data["results"][0]["now"]["temperature"],
                "description": data["results"][0]["now"]["text"],
            }
            return weather
        else:
            raiseException(
                f"Failed to retrieve weather: {response.status_code}")

    def _run(self, para: str) -> str:
        return self.get_weather(para)

2.weather.yaml file
The contents of the weather.yaml file are as follows:

name: weather
description: Search the current weather of a city
parameters:
  type: object
  properties:
    city:
      type: string
      description: city name
  required:
    - city

2. Customized weather query tool call
Customize the weather query tool call and import the Weather tool in main.py. As follows:

run_tool([Weather()], llm, [
    "How is the weather in Beijing today?",
    "What's the weather like in Shanghai today",
])

Among them, the run_tool() function is implemented as follows:

def run_tool(tools, llm, prompt_chain: List[str]):
    loaded_tolls = [] # Tools for storing loads
    for tool in tools: # Load tools one by one
        if isinstance(tool, str):
            loaded_tolls.append(load_tools([tool], llm=llm)[0]) # load_tools returns a list
        else:
            loaded_tolls.append(tool) # If it is a custom tool, add it directly to the list
    agent = initialize_agent( # Initialize agent
        loaded_tolls, llm,
        agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, # agent type: agent using structured chat
        verbose=True,
        handle_parsing_errors=True
    )
    for prompt in prompt_chain: # Enter prompts one by one
        agent.run(prompt)

1.load_tools() function
Load the corresponding tool according to the tool name, as shown below:

def load_tools(
    tool_names: List[str],
    llm: Optional[BaseLanguageModel] = None,
    callbacks: Callbacks = None,
    **kwargs: Any,
) -> List[BaseTool]:

2.initialize_agent() function
Load an agent executor based on the tool list and LLM, as follows:

def initialize_agent(
    tools: Sequence[BaseTool],
    llm: BaseLanguageModel,
    agent: Optional[AgentType] = None,
    callback_manager: Optional[BaseCallbackManager] = None,
    agent_path: Optional[str] = None,
    agent_kwargs: Optional[dict] = None,
    *,
    tags: Optional[Sequence[str]] = None,
    **kwargs: Any,
) -> AgentExecutor:

Among them, the agent defaults to AgentType.ZERO_SHOT_REACT_DESCRIPTION. The one used in this article is AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, a zero-shot react agent optimized for chat models that can call tools with multiple inputs.
3.run() function
Convenience method for execution chains. The main difference between this method and Chain.__call__ is that this method expects input to be passed directly as positional or keyword arguments, while Chain.__call__ Expects a single input dictionary containing all inputs. As follows:

def run(
    self,
    *args: Any,
    callbacks: Callbacks = None,
    tags: Optional[List[str]] = None,
    metadata: Optional[Dict[str, Any]] = None,
    **kwargs: Any,
) -> Any:


4. Result analysis
The result output is as follows:

> Entering new AgentExecutor chain...
======
======

Action:
``
{"action": "weather", "action_input": "Beijing"}
``
Observation: {'temperature': '20', 'description': 'clear'}
Thought:======
======

Action:
``
{"action": "Final Answer", "action_input": "According to the query results, the weather in Beijing today is sunny and the temperature is 20℃."}
``

> Finished chain.


> Entering new AgentExecutor chain...
======
======

Action:
``
{"action": "weather", "action_input": "Shanghai"}
``
Observation: {'temperature': '20', 'description': 'clear'}
Thought:======
======

Action:
``
{"action": "Final Answer", "action_input": "According to the latest weather data, the weather conditions in Shanghai today are sunny and the temperature is 20℃."}
``

> Finished chain.

At the beginning, I couldn’t find the place to identify the entity city. Later, when I debugged ChatGLM3/langchain_demo/ChatGLM3.py->_call(), I found a huge prompt. Isn’t this a zero-prompt? (AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION)? By the way, the code of LangChain is really difficult to debug.

3. Registration tools enhance LLM capabilities
1. Registration tool
The capabilities of the model can be enhanced by registering new tools in tool_registry.py. Just use the @register_tool decoration function to complete the registration. For tool declarations, the function name is the name of the tool, and the function docstring is the description of the tool; for tool parameters, use Annotated[typ: type, description: str, required: bool] to mark the type, description, and whether the parameters are required. Register the get_weather() function as follows:

@register_tool
def get_weather( # Utility function
        city_name: Annotated[str, 'The name of the city to be queried', True],
) -> str:
    """
    Get the current weather for `city_name`
    """

    if not isinstance(city_name, str): # Parameter type check
        raise TypeError("City name must be a string")

    key_selection = { # selected key
        "current_condition": ["temp_C", "FeelsLikeC", "humidity", "weatherDesc", "observation_time"],
    }
    import requests
    try:
        resp = requests.get(f"https://wttr.in/{city_name}?format=j1")
        resp.raise_for_status()
        resp = resp.json()
        ret = {k: {_v: resp[k][0][_v] for _v in v} for k, v in key_selection.items()}
    except:
        import traceback
        ret = "Error encountered while fetching weather data!\\
" + traceback.format_exc()

    return str(ret)

The specific tool registration implementation method @register_tool decoration function is as follows:

def register_tool(func: callable): # Register tool
    tool_name = func.__name__ # Tool name
    tool_description = inspect.getdoc(func).strip() # Tool description
    python_params = inspect.signature(func).parameters # Tool parameters
    tool_params = [] # tool parameter description
    for name, param in python_params.items(): # Traverse parameters
        annotation = param.annotation # Parameter annotation
        if annotation is inspect.Parameter.empty:
            raise TypeError(f"Parameter `{name}` missing type annotation") # The parameter is missing annotation
        if get_origin(annotation) != Annotated: # Parameter annotation is not Annotated
            raise TypeError(f"Annotation type for `{name}` must be typing.Annotated") # Parameter annotation must be Annotated

        typ, (description, required) = annotation.__origin__, annotation.__metadata__ # Parameter type, parameter description, whether it is required
        typ: str = str(typ) if isinstance(typ, GenericAlias) else typ.__name__ # Parameter type name
        if not isinstance(description, str): # Parameter description must be a string
            raise TypeError(f"Description for `{name}` must be a string")
        if not isinstance(required, bool): # Whether it must be a Boolean value
            raise TypeError(f"Required for `{name}` must be a bool")

        tool_params.append({ #Add parameter description
            "name": name,
            "description": description,
            "type": typ,
            "required": required
        })
    tool_def = { # tool definition
        "name": tool_name,
        "description": tool_description,
        "params": tool_params
    }

    print("[registered tool] " + pformat(tool_def)) # Print tool definition
    _TOOL_HOOKS[tool_name] = func # Register tool
    _TOOL_DESCRIPTIONS[tool_name] = tool_def # Add tool definition

    return func

2. Call tools
Reference file ChatGLM3/tool_using/openai_api_demo.py, as shown below:

def main():
    messages = [ #Conversation messages
        system_info,
        {
            "role": "user",
            "content": "Help me check the weather in Beijing",
        }
    ]
    response = openai.ChatCompletion.create( # Call OpenAI API
        model="chatglm3",
        messages=messages,
        temperature=0,
        return_function_call=True
    )
    function_call = json.loads(response.choices[0].message.content) # Get function call information
    logger.info(f"Function Call Response: {function_call}") # Print function call information

    tool_response = dispatch_tool(function_call["name"], function_call["parameters"]) # Call function
    logger.info(f"Tool Call Response: {tool_response}") # Print function call results

    messages = response.choices[0].history # Get historical conversation information
    messages.append(
        {
            "role": "observation",
            "content": tool_response, # Call the function to return the result
        }
    )

    response = openai.ChatCompletion.create( # Call OpenAI API
        model="chatglm3",
        messages=messages,
        temperature=0,
    )
    logger.info(response.choices[0].message.content) #Print the conversation results

References:
[1]https://github.com/THUDM/ChatGLM3/tree/main
[2]https://python.langchain.com/docs/integrations/tools