导入工具包
rom langchain_core.tools import BaseTool
from typing import Sequence, Optional, List
from langchain_core.prompts import BasePromptTemplate
import re
from langchain_core.tools import tool
from langchain_core.prompts.chat import (ChatPromptTemplate,HumanMessagePromptTemplate,SystemMessagePromptTemplate,
)
from langchain.chains.llm import LLMChain
from langchain_openai import ChatOpenAI
langchain初始化智能体源码中prompt
PREFIX = 'Respond to the human as helpfully and accurately as possible. You have access to the following tools:'
SUFFIX = 'Begin! Reminder to ALWAYS respond with a valid json blob of a single action. Use tools if necessary. Respond directly if appropriate. Format is Action:```$JSON_BLOB```then Observation:.\nThought:'
HUMAN_MESSAGE_TEMPLATE = '''{input}{agent_scratchpad}'''
FORMAT_INSTRUCTIONS = '''Use a json blob to specify a tool by providing an action key (tool name) and an action_input key (tool input).Valid "action" values: "Final Answer" or {tool_names}Provide only ONE action per $JSON_BLOB, as shown:{{{{
“action”: $TOOL_NAME,
“action_input”: $INPUT
}}}}
Follow this format:Question: input question to answer
Thought: consider previous and subsequent steps
Action:
$JSON_BLOB
Observation: action result
... (repeat Thought/Action/Observation N times)
Thought: I know what to respond
Action:
{{{{
“action”: “Final Answer”,
“action_input”: “Final response to human”
}}}}
prompt生成函数
def create_prompt(tools: Sequence[BaseTool],prefix: str = PREFIX,suffix: str = SUFFIX,human_message_template: str = HUMAN_MESSAGE_TEMPLATE,format_instructions: str = FORMAT_INSTRUCTIONS,input_variables: Optional[List[str]] = None,memory_prompts: Optional[List[BasePromptTemplate]] = None,
) -> BasePromptTemplate:tool_strings = []for tool in tools:args_schema = re.sub("}", "}}", re.sub("{", "{{", str(tool.args)))tool_strings.append(f"{tool.name}: {tool.description}, args: {args_schema}")formatted_tools = "\n".join(tool_strings)tool_names = ", ".join([tool.name for tool in tools])format_instructions = format_instructions.format(tool_names=tool_names)template = "\n\n".join([prefix, formatted_tools, format_instructions, suffix])if input_variables is None:input_variables = ["input", "agent_scratchpad"]_memory_prompts = memory_prompts or []messages = [SystemMessagePromptTemplate.from_template(template),*_memory_prompts,HumanMessagePromptTemplate.from_template(human_message_template),]return ChatPromptTemplate(input_variables=input_variables, messages=messages) # type: ignore[arg-type]
工具定义
@tool
def multiply(first_int: int, second_int: int) -> int:"""将两个整数相乘。"""print('---------multiply-----------------')return first_int * second_int@tool
def add(first_int: int, second_int: int) -> int:"将两个整数相加。"print('---------add-----------------')return first_int + second_int@tool
def exponentiate(base: int, exponent: int) -> int:"指数运算"print('---------exponentiate-----------------')with open('小黑黑.txt', 'w', encoding='utf-8') as f:f.write('小黑黑')return base**exponent
大语言模型接口初始化
zhipu_key = 'a66c6fc7748xxxxxxxxxxxxxxxx7ctC83zWJo'
llm = ChatOpenAI(temperature=0.01,model="glm-4-flash",openai_api_key=zhipu_key,openai_api_base="https://open.bigmodel.cn/api/paas/v4/"
)
定义工作流
tools = [multiply, add, exponentiate]
prompt = create_prompt(tools=tools)
llm = ChatOpenAI(temperature=0.01,model="glm-4-flash",openai_api_key=zhipu_key,openai_api_base="https://open.bigmodel.cn/api/paas/v4/"
)
定义智能体
from langchain.agents import StructuredChatAgent
from langchain.agents.structured_chat.output_parser import StructuredChatOutputParserWithRetries
# 定义智能体
structuredChatAgent = StructuredChatAgent(llm_chain=llm_chain,allowed_tools=[tool.name for tool in tools],output_parser=StructuredChatOutputParserWithRetries())
运行智能体
from langchain.agents.agent import AgentExecutor
# 执行智能体
excuter = AgentExecutor.from_agent_and_tools(agent=structuredChatAgent,tools=tools,callback_manager=None,verbose=True)excuter.invoke("调用api计算3加5乘2等于多少?")
