LLM大语言模型（十五）：LangChain的Agent中使用自定义的ChatGLM，且底层调用的是remote的ChatGLM3-6B的HTTP服务_langchain调用本地chatglm

背景

本文搭建了一个完整的LangChain的Agent，调用本地启动的ChatGLM3-6B的HTTP server。

为后续的RAG做好了准备。

增加服务端role：observation

ChatGLM3的官方demo：openai_api_demo目录

api_server.py文件

class ChatMessage(BaseModel):
    # role: Literal["user", "assistant", "system", "function"]
    role: Literal["user", "assistant", "system", "function","observation"]
    content: str = None
    name: Optional[str] = None
    function_call: Optional[FunctionCallResponse] = None

修改role列表，增加了“observation”。

这是因为LangChain的Agent执行过程，是ReAct模式，在执行完tool调用后，会生成一个observation角色的消息。

在将LangChain的prompt转换为ChatGLM3的prompt时，也保留了observation角色，但是在服务启动时，接口允许的role却没有observation，会导致接口调用失败。

ChatGLM3-6B 本地HTTP服务启动

参考：

LLM大语言模型（一）：ChatGLM3-6B本地部署_llm3 部署-CSDN博客

自定义LLM

自定义LLM内部访问的是HTTP server。

将LangChain Agent的prompt转换为ChatGLM3能识别的prompt。

prompt转换参考：LLM大语言模型（十三）：ChatGLM3-6B兼容Langchain的Function Call的一步一步的详细转换过程记录_langchain+chatglm3-CSDN博客

import ast
import requests
import json
from typing import Any, List, Optional
from langchain.llms.base import LLM
from langchain_core.callbacks import CallbackManagerForLLMRun
from output_parse import getFirstMsg,parse_tool
 
 
class MyChatGLM(LLM):
    max_token: int = 8192
    # do_sample: bool = False
    do_sample: bool = True
    temperature: float = 0.8
    top_p = 0.8
    tokenizer: object = None
    model: object = None
    history: List = []
    has_search: bool = False
    model_name: str = "chatglm3-6b"
    url: str = "http://localhost:8000/v1/chat/completions"
    tools: List = []
 
    # def __init__(self):
    #     super().__init__()
 
    @property
    def _llm_type(self) -> str:
        return "MyChatGLM"
 
 
    def _tool_history(self, prompt: str):
            ans = []
 
            tool_prompts = prompt.split(
                "You have access to the following tools:\n\n")[1].split("\n\nUse a json blob")[0].split("\n")
            tools_json = []
 
            for tool_desc in tool_prompts:
                name = tool_desc.split(":")[0]
                description = tool_desc.split(", args:")[0].split(":")[1].strip()
                parameters_str = tool_desc.split("args:")[1].strip()
                parameters_dict = ast.literal_eval(parameters_str)
                params_cleaned = {}
                for param, details in parameters_dict.items():
                    params_cleaned[param] = {'description': details['description'], 'type': details['type']}
 
                tools_json.append({
                    "name": name,
                    "description": description,
                    "parameters": params_cleaned
                })
 
            ans.append({
                "role": "system",
                "content": "Answer the following questions as best as you can. You have access to the following tools:",
                "tools": tools_json
            })
 
            dialog_parts = prompt.split("Human: ")
            for part in dialog_parts[1:]:
                if "\nAI: " in part:
                    user_input, ai_response = part.split("\nAI: ")
                    ai_response = ai_response.split("\n")[0]
                else:
                    user_input = part
                    ai_response = None
 
                ans.append({"role": "user", "content": user_input.strip()})
                if ai_response:
                    ans.append({"role": "assistant", "content": ai_response.strip()})
 
            query = dialog_parts[-1].split("\n")[0]
            return ans, query
 
    def _extract_observation(self, prompt: str):
        return_json = prompt.split("Observation: ")[-1].split("\nThought:")[0]
        self.history.append({
            "role": "observation",
            "content": return_json
        })
        return
 
    def _extract_tool(self):
        if len(self.history[-1]["metadata"]) > 0:
            metadata = self.history[-1]["metadata"]
            content = self.history[-1]["content"]
 
            lines = content.split('\n')
            for line in lines:
                if 'tool_call(' in line and ')' in line and self.has_search is False:
                    # 获取括号内的字符串
                    params_str = line.split('tool_call(')[-1].split(')')[0]
 
                    # 解析参数对
                    params_pairs = [param.split("=") for param in params_str.split(",") if "=" in param]
                    params = {pair[0].strip(): pair[1].strip().strip("'\"") for pair in params_pairs}
                    action_json = {
                        "action": metadata,
                        "action_input": params
                    }
                    self.has_search = True
                    print("*****Action*****")
                    print(action_json)
                    print("*****Answer*****")
                    return f"""
Action: 
```
{json.dumps(action_json, ensure_ascii=False)}
```"""
        final_answer_json = {
            "action": "Final Answer",
            "action_input": self.history[-1]["content"]
        }
        self.has_search = False
        return f"""
Action: 
```
{json.dumps(final_answer_json, ensure_ascii=False)}
```"""
 
    def _call(self, prompt: str, history: List = [], stop: Optional[List[str]] = ["<|user|>"]):
        if not self.has_search:
            self.history, query = self._tool_history(prompt)
            if self.history[0]:
                self.tools = self.history[0]["tools"]
        else:
            self._extract_observation(prompt)
            query = ""
        print(self.history)
        data = {}
        data["model"] = self.model_name
        data["messages"] = self.history
        data["temperature"] = self.temperature
        data["max_tokens"] = self.max_token
        data["tools"] = self.tools
        resp = self.doRequest(data)
        
        msg = {}
        respjson = json.loads(resp)
        if respjson["choices"]:
            if respjson["choices"][0]["finish_reason"] == 'function_call':
                msg["metadata"] = respjson["choices"][0]["message"]["function_call"]["name"]
            else:
                msg["metadata"] = ''
            msg["role"] = "assistant"
            msg["content"] = respjson["choices"][0]["message"]["content"]
 
        self.history.append(msg)
        print(self.history)
        response = self._extract_tool()
        history.append((prompt, response))
        return response
 
 
    def doRequest(self,payload:dict) -> str:
        # 请求头
        headers = {"content-type":"application/json"}
        # json形式，参数用json
        res = requests.post(self.url,json=payload,headers=headers)
        return res.text

定义tool

使用LangChain中Tool的方式：继承BaseTool

Tool实现方式对prompt的影响，参考：LLM大语言模型（十四）：LangChain中Tool的不同定义方式，对prompt的影响-CSDN博客

class WeatherInput(BaseModel):
    location: str = Field(description="the location need to check the weather")
 
 
class Weather(BaseTool):
    name = "weather"
    description = "Use for searching weather at a specific location"
    args_schema: Type[BaseModel] = WeatherInput
 
    def __init__(self):
        super().__init__()
 
    def _run(self, location: str) -> dict[str, Any]:
        weather = {
            "temperature": "20度",
            "description": "温度适中",
        }
        return weather

LangChain Agent调用

设置Agent使用了2个tool：Calculator() Weather()，看是否能正确调用。

    # Get the prompt to use - you can modify this!
    prompt = hub.pull("hwchase17/structured-chat-agent")
    prompt.pretty_print()
    tools = [Calculator(),Weather()]
    # Choose the LLM that will drive the agent
    # Only certain models support this
    # Choose the LLM to use
    llm = MyChatGLM()
 
    # Construct the agent
    agent = create_structured_chat_agent(llm, tools, prompt)
    # Create an agent executor by passing in the agent and tools
    agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, handle_parsing_errors=True)
 
    ans = agent_executor.invoke({"input": "北京天气怎么样？"})
    print(ans)

调用结果：

> Entering new AgentExecutor chain...
[{'role': 'system', 'content': 'Answer the following questions as best as you can. You have access to the following tools:', 'tools': [{'name': 'Calculator', 'description': 'Useful for when you need to calculate math problems', 'parameters': {'calculation': {'description': 'calculation to perform', 'type': 'string'}}}, {'name': 'weather', 'description': 'Use for searching weather at a specific location', 'parameters': {'location': {'description': 'the location need to check the weather', 'type': 'string'}}}]}, {'role': 'user', 'content': '北京天气怎么样？\n\n\n (reminder to respond in a JSON blob no matter what)'}]
[{'role': 'system', 'content': 'Answer the following questions as best as you can. You have access to the following tools:', 'tools': [{'name': 'Calculator', 'description': 'Useful for when you need to calculate math problems', 'parameters': {'calculation': {'description': 'calculation to perform', 'type': 'string'}}}, {'name': 'weather', 'description': 'Use for searching weather at a specific location', 'parameters': {'location': {'description': 'the location need to check the weather', 'type': 'string'}}}]}, {'role': 'user', 'content': '北京天气怎么样？\n\n\n (reminder to respond in a JSON blob no matter what)'}, {'metadata': 'weather', 'role': 'assistant', 'content': "weather\n ```python\ntool_call(location='北京')\n```"}]
*****Action*****
{'action': 'weather', 'action_input': {'location': '北京'}}
*****Answer*****

Action:
```
{"action": "weather", "action_input": {"location": "北京"}}
```{'temperature': '20度', 'description': '温度适中'}

[{'role': 'system', 'content': 'Answer the following questions as best as you can. You have access to the following tools:', 'tools': [{'name': 'Calculator', 'description': 'Useful for when you need to calculate math problems', 'parameters': {'calculation': {'description': 'calculation to perform', 'type': 'string'}}}, {'name': 'weather', 'description': 'Use for searching weather at a specific location', 'parameters': {'location': {'description': 'the location need to check the weather', 'type': 'string'}}}]}, {'role': 'user', 'content': '北京天气怎么样？\n\n\n (reminder to respond in a JSON blob no matter what)'}, {'metadata': 'weather', 'role': 'assistant', 'content': "weather\n ```python\ntool_call(location='北京')\n```"}, {'role': 'observation', 'content': "{'temperature': '20度', 'description': '温度适中'}"}]
[{'role': 'system', 'content': 'Answer the following questions as best as you can. You have access to the following tools:', 'tools': [{'name': 'Calculator', 'description': 'Useful for when you need to calculate math problems', 'parameters': {'calculation': {'description': 'calculation to perform', 'type': 'string'}}}, {'name': 'weather', 'description': 'Use for searching weather at a specific location', 'parameters': {'location': {'description': 'the location need to check the weather', 'type': 'string'}}}]}, {'role': 'user', 'content': '北京天气怎么样？\n\n\n (reminder to respond in a JSON blob no matter what)'}, {'metadata': 'weather', 'role': 'assistant', 'content': "weather\n ```python\ntool_call(location='北京')\n```"}, {'role': 'observation', 'content': "{'temperature': '20度', 'description': '温度适中'}"}, {'metadata': '', 'role': 'assistant', 'content': '根据最新的气象数据 
，北京的天气情况如下：温度为20度，天气状况适中。'}]

Action:
```
{"action": "Final Answer", "action_input": "根据最新的气象数据，北京的天气情况如下：温度为20度，天气状况适中。"}
```

> Finished chain.
{'input': '北京天气怎么样？', 'output': '根据最新的气象数据，北京的天气情况如下：温度为20度，天气状况适中。'}