当前位置：首页 > news >正文

小黑一步步探索大模型应用：langchain中AgentExecutor的call方法初探demo（智能体调用）

news 来源：原创 2025/6/5 15:52:29

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”
}}}}

导包和大模型定义

from 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 ChatOpenAIdef 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**exponenttools = [multiply, add, exponentiate]
prompt = create_prompt(tools=tools)
zhipu_key = 'a111111111111111111111111111111'
llm = ChatOpenAI(temperature=0.01,model="glm-4-flash",openai_api_key=zhipu_key,openai_api_base="https://open.bigmodel.cn/api/paas/v4/"
)

定义chain

# 定义chain
llm_chain = LLMChain(llm=llm,prompt=prompt,callback_manager=None,
)
llm_chain.invoke({'input':'调用api计算3加5乘2等于多少?', 'agent_scratchpad':''})

{‘input’: ‘调用api计算3加5乘2等于多少?’,
‘agent_scratchpad’: ‘’,
‘text’: ‘Thought: The user wants to calculate 3 + 5 * 2. According to the order of operations, multiplication should be performed before addition. So, I will first multiply 5 by 2 and then add 3 to the result.\nAction:\njson\n{\n "action": "multiply",\n "action_input": {\n "first_int": 5,\n "second_int": 2\n }\n}\n’}

定义智能体

from langchain.agents.agent import AgentExecutor
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())

AgentExecutor类的演示，执行智能体（在这里通过继承AgentExecutor演示call方法）

import time
class XiaoHeiAgentExecutor(AgentExecutor):def xiaohei_call(self, inputs, run_manager=None):"""Run text through and get agent response."""# Construct a mapping of tool name to tool for easy lookup# 工具列表name_to_tool_map = {tool.name: tool for tool in self.tools}# We construct a mapping from each tool to a color, used for logging.# 不同工具结果对应的显示颜色color_mapping = get_color_mapping([tool.name for tool in self.tools], excluded_colors=["green", "red"])# 中间的步骤intermediate_steps: List[Tuple[AgentAction, str]] = []# Let's start tracking the number of iterations and time elapsediterations = 0time_elapsed = 0.0start_time = time.time()# We now enter the agent loop (until it returns something).# *********************************开始迭代执行智能体，直到AgentFinish状态（或者达到最大迭代次数）*********************************while self._should_continue(iterations, time_elapsed):# 对工具进行选择性使用，并返回这一轮的输出next_step_output = self._take_next_step(name_to_tool_map,color_mapping,inputs,intermediate_steps,run_manager=run_manager,)# 判断该输出中是否存在结束标志，如果跳出并返回最终结果if isinstance(next_step_output, AgentFinish):return self._return(next_step_output, intermediate_steps, run_manager=run_manager)# 记录中间步骤intermediate_steps.extend(next_step_output)if len(next_step_output) == 1:next_step_action = next_step_output[0]# See if tool should return directlytool_return = self._get_tool_return(next_step_action)if tool_return is not None:return self._return(tool_return, intermediate_steps, run_manager=run_manager)iterations += 1time_elapsed = time.time() - start_time# 超过了最大迭代次数，直接返回output = self._action_agent.return_stopped_response(self.early_stopping_method, intermediate_steps, **inputs)return self._return(output, intermediate_steps, run_manager=run_manager)
xiaohei_agent_executor = XiaoHeiAgentExecutor.from_agent_and_tools(agent=structuredChatAgent,tools=tools,callback_manager=None,verbose=True)
xiaohei_agent_executor.xiaohei_call({'input':'调用api计算3加5乘2等于多少?', 'agent_scratchpad':''})