背景
本文搭建了一个完整的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 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 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度，天气状况适中。'}
