小黑一层层削苹果皮式大模型应用探索：langchain中智能体思考和执行工具的demo

引言

小黑黑通过探索langchain源码，设计了一个关于agent使用工具的一个简化版小demo（代码可以跑通），主要流程：
1.问题输入给大模型。
2.大模型进行思考，输出需要执行的action和相关思考信息。
3.通过代理（demo中为perform_action函数）执行action进行工具调用，并返回工具调用结果（res）。
4.将action和res进行记录，继续喂给大模型进行思考，让大模型输出下一步action，以此类推…

demo准备工作

导包

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.agents import StructuredChatAgent
from langchain.chains.llm import LLMChain
from langchain_openai import ChatOpenAI
# 导入智能体
from langchain.agents.agent import Agent
from langchain.agents.structured_chat.output_parser import StructuredChatOutputParserWithRetries
from pydantic import Field
from langchain.agents.agent import Agent, AgentOutputParser
from typing import Any, List, Optional, Sequence, Tuple, Union
from langchain_core.agents import AgentAction
from langchain_core.language_models import BaseLanguageModel
from langchain_core.callbacks import BaseCallbackManager
from langchain_core.agents import AgentAction, AgentFinish, AgentStep
from langchain.agents.tools import InvalidTool

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

大模型定义（key脱敏处理，可以免费申请）

zhipu_key = '111111111111111111111111111111111111111111111'
llm = ChatOpenAI(
    temperature=0.01,
    model="glm-4-flash",
    openai_api_key=zhipu_key,
    openai_api_base="https://open.bigmodel.cn/api/paas/v4/"
)

组合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

tools = [multiply, add, exponentiate]
prompt = create_prompt(tools=tools)

定义chain和智能体

# 定义chain
llm_chain = LLMChain(
    llm=llm,
    prompt=prompt,
    callback_manager=None,
)
# 定义智能体
structuredChatAgent = StructuredChatAgent(
            llm_chain=llm_chain,
            allowed_tools=[tool.name for tool in tools],
            output_parser=StructuredChatOutputParserWithRetries())

demo核心代码

代理函数

# 执行智能体
def perform_action(agent, name_to_tool_map, agent_action, run_manager=None):
    # 如果选择的工具在工具箱里
    if agent_action.tool in name_to_tool_map:
        # 获取所有工具
        tool = name_to_tool_map[agent_action.tool]
        # 判断是否直接返回工具
        return_direct = tool.return_direct
        # 获取智能体的输出前缀，例如思考内容用“Thought”,观察内容用“Observation:”，{'llm_prefix': 'Thought:', 'observation_prefix': 'Observation: '}
        tool_run_kwargs = agent.tool_run_logging_kwargs()
        # 执行工具
        observation = tool.run(
                agent_action.tool_input,
                verbose=None,
                color=None,
                callbacks=run_manager.get_child() if run_manager else None,
                **tool_run_kwargs,
            )
    # 选择的工具不在工具箱内，返回“无效工具相关信息给大模型”
    else:
        tool_run_kwargs = agent.tool_run_logging_kwargs()
        observation = InvalidTool().run(
            {
                "requested_tool_name": agent_action.tool,
                "available_tool_names": list(name_to_tool_map.keys()),
            },
            verbose=self.verbose,
            color=None,
            callbacks=run_manager.get_child() if run_manager else None,
             **tool_run_kwargs,
            )
    # 以AgentStep的形式返回
    return AgentStep(action=agent_action, observation=observation)

大模型通过思考，一步步调用工具

# 历史动作和工具调用结果
intermediate_steps=[]
# 输入问题
question = '调用api计算3加5乘2等于多少?'
# 如果action为AgentFinish,则为最后一步，跳出循环
action = None
# 如果action为AgentFinish,则为最后一步，跳出循环
while True:
    # 大语言模型通过思考给出action
    action = structuredChatAgent.plan(intermediate_steps=intermediate_steps, input=question)
    print('[[[action]]]:', action)
    # 跳出循环
    if isinstance(action, AgentFinish):
        print('执行结束，得到答案')
        break
    # 通过 代理 执行action
    agent_step = perform_action(structuredChatAgent, dict((tool.name,tool) for tool in tools), action)
    print('[[[agent_step]]]:', agent_step)
    # 将action和执行结果保存
    intermediate_steps.append((agent_step.action, agent_step.observation))
    print('*****************************************************************************************')

程序运行结果：
[[[action]]]: tool=‘multiply’ tool_input={‘first_int’: 5, ‘second_int’: 2} log=‘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’
---------multiply-----------------
[[[agent_step]]]: action=AgentAction(tool=‘multiply’, tool_input={‘first_int’: 5, ‘second_int’: 2}, log=‘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’) observation=10

[[[action]]]: tool=‘add’ tool_input={‘first_int’: 3, ‘second_int’: 10} log=‘Action:\njson\n{\n "action": "add",\n "action_input": {\n "first_int": 3,\n "second_int": 10\n }\n}\n’
---------add-----------------
[[[agent_step]]]: action=AgentAction(tool=‘add’, tool_input={‘first_int’: 3, ‘second_int’: 10}, log=‘Action:\njson\n{\n "action": "add",\n "action_input": {\n "first_int": 3,\n "second_int": 10\n }\n}\n’) observation=13

[[[action]]]: return_values={‘output’: ‘13’} log=‘Action:\njson\n{\n "action": "Final Answer",\n "action_input": "13"\n}\n’
执行结束，得到答案