背景
本文搭建了一个完整的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 = Nonename: Optional[str] = Nonefunction_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_toolclass MyChatGLM(LLM):max_token: int = 8192# do_sample: bool = Falsedo_sample: bool = Truetemperature: float = 0.8top_p = 0.8tokenizer: object = Nonemodel: object = Nonehistory: List = []has_search: bool = Falsemodel_name: str = "chatglm3-6b"url: str = "http://localhost:8000/v1/chat/completions"tools: List = []# def __init__(self):# super().__init__()@propertydef _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 = partai_response = Noneans.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, querydef _extract_observation(self, prompt: str):return_json = prompt.split("Observation: ")[-1].split("\nThought:")[0]self.history.append({"role": "observation","content": return_json})returndef _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 = Trueprint("*****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 = Falsereturn 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_namedata["messages"] = self.historydata["temperature"] = self.temperaturedata["max_tokens"] = self.max_tokendata["tools"] = self.toolsresp = 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 responsedef doRequest(self,payload:dict) -> str:# 请求头headers = {"content-type":"application/json"}# json形式，参数用jsonres = 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] = WeatherInputdef __init__(self):super().__init__()def _run(self, location: str) -> dict[str, Any]:weather = {"temperature": "20度","description": "温度适中",}return weatherLangChain 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 usellm = MyChatGLM()# Construct the agentagent = create_structured_chat_agent(llm, tools, prompt)# Create an agent executor by passing in the agent and toolsagent_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度，天气状况适中。'}
![[wp]第一届 “帕鲁杯“ --应急响应](https://img-blog.csdnimg.cn/direct/137420103be54ef4a771719cf4247a6a.png)
