LangChain 从入门到实战 -六大组件之Agent_langchain agent_langchain的agent，零基础入门到精通，收藏这篇就够了-开发者社区

一、代理原理

代理的核心思想是使用语言模型来选择一系列操作。在链中,一系列操作是硬编码的。在代理中,语言模型被用作推理引擎来确定要采取哪些操作以及按什么顺序。
需要了解的重要概念:

AgentAction:表示代理应采取的操作。它有一个tool属性(要调用的工具的名称)和一个tool_input属性(该工具的输入)
AgentFinish:表示代理已完成,应返回给用户。它有一个return_values参数,它通常只有一个键——output,即一个字符串,因此通常只返回这个键。
intermediate_steps:代表之前的代理操作和相应输出。这对于传递给未来迭代非常重要,所以代理知道它已经完成了哪些工作。它被编码为List[Tuple[AgentAction, Any]].

使用语言模型来选择一系列操作
AgentAction表示一个操作 [] （messagelog）
AgentFinish表示代理已完成 [AgentFinish(return_values={‘output’: ‘单词"color"有5个字母。’}, log=‘单词"color"有5个字母。’)]
intermediate_steps记录之前的操作和输出,用于未来迭代【tool3 ，tools2，tools3】

Agent（代理）：在 Langchain 中，Agent 是指一个智能合约的实例。它代表了合约的执行者，可以看作是一个具有特定功能的智能合约的化身。Agent 能够根据合约的逻辑执行相应的操作，如转账、查询余额等。它是区块链网络中的一个基本单位，可以理解为一个可编程的、自主执行任务的角色。

Chain（链）：Langchain 中的 Chain 指的是一个由多个 Agent 组成的序列。这些 Agent 按照一定的顺序链接在一起，形成一个链式结构。Chain 是 Langchain 平台的基本组织结构，用于实现分布式账本、智能合约等功能。每个 Chain 都对应一个唯一的标识符，如一个哈希值。Chain 中的 Agent 按照预先设定的规则执行任务，共同维护整个链的稳定和安全。

Agent 是 Chain 的基本组成单元，一个 Chain 由多个 Agent 构成。

Chain 是 Agent 存在的载体，Agent 通过加入 Chain 来实现其功能和价值。

Agent 和 Chain 共同构成了 Langchain 平台的核心技术体系

1、源码查看

https://github.com/langchain-ai/langchain/blob/804390ba4bcc306b90cb6d75b7f01a4231ab6463/libs/langchain/langchain/agents/init.py

2、创建工具（装饰器方法）

from langchain.chat_models import ChatOpenAI # 创建语言模型 llm = ChatOpenAI(temperature=0) # 构建工具 from langchain.agents import tool @tool def get_word_length(word: str) -> int: """返回单词的长度""" return len(word) # 创建工具清单 tools = [get_word_length]

看看工具函数
一个name参数，代表这个函数名称
一个description，返回这个工具的使用说明

3、绑定工具

# 绑定工具 from langchain.tools.render import format_tool_to_openai_function # 格式化函数作为符合openai格式的描述体 llm_with_tools = llm.bind( functions=[format_tool_to_openai_function(t) for t in tools] )

format_tool_to_openai_function 这个方法将工具函数，转换为符合 OpenAi格式的描述体，即fucntion calling

函数调用是构建由 LLM 驱动的聊天机器人或代理的一项重要功能，这些聊天机器人或代理需要检索 LLM 的上下文或通过将自然语言转换为 API 调用来与外部工具交互。

4、创建代理

下面代码，一个是prompt，创建了2个输入key

input 用于用户输入内容
agent_scratchpad 用户代理中间使用过程的记录

# 创建代理 from langchain.agents.format_scratchpad import format_to_openai_functions from langchain.agents.output_parsers import OpenAIFunctionsAgentOutputParser from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个非常优秀的人，但是不会计算单词的长度"), ("user", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad"), # 消息体当中添加一个 key，记录代理的使用过程 ]) agent = { "input": lambda x: x["input"], "agent_scratchpad": lambda x: format_to_openai_functions(x['intermediate_steps']) # 代理的使用过程 } | prompt | llm_with_tools | OpenAIFunctionsAgentOutputParser()

现在来使用它

output = agent.invoke({ "input": "关于color有多少个字母?", "intermediate_steps": [],# 中间步骤 })

可以看到生成了一个AgentActionMessageLog，这个表示代理应该要采集的动作，并没有到结束

为了更加直观，我复制出来
可以看到，将用户的问：“关于color有多少个字母?”这个问题去，提取了 color，并且准备调用get_word_length工具函数
关键字：tool、tool_input

现在我询问另外一个问题，看看输出的样式，直接为AgentFinish，这个表明代理已经完成动作，原因在于，我们封装的工具函数，并没有解决 “太阳和地球的距离”

关键字：return_values

5、完整代码

好了，到现在是不是看起来有点懵逼，现在，通过下面代码来加深一步过程

一个代理Agent，从白话文来说-简化成2个过程

AgentAction 代理开始
AgentFinish 代理结束，当只有看到这个类型才视为结束

那么现在，将前面的代码抽取出来，通过while True 完善工具链的使用

from langchain.chat_models import ChatOpenAI # 创建语言模型 llm = ChatOpenAI(temperature=0) # 构建工具 from langchain.agents import tool @tool def get_word_length(word: str) -> int: """返回单词的长度""" return len(word) # 创建工具清单 tools = [get_word_length] # 绑定工具 from langchain.tools.render import format_tool_to_openai_function # 格式化函数作为符合openai格式的描述体 llm_with_tools = llm.bind( functions=[format_tool_to_openai_function(t) for t in tools] ) # 创建代理 from langchain.agents.format_scratchpad import format_to_openai_functions from langchain.agents.output_parsers import OpenAIFunctionsAgentOutputParser from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个非常优秀的人，但是不会计算单词的长度"), ("user", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad"), # 消息体当中添加一个 key，记录代理的使用过程 ]) agent = { "input": lambda x: x["input"], "agent_scratchpad": lambda x: format_to_openai_functions(x['intermediate_steps']) # 代理的使用过程 } | prompt | llm_with_tools | OpenAIFunctionsAgentOutputParser() from langchain.schema.agent import AgentFinish intermediate_steps = [] count = 1 while True: output = agent.invoke({ "input": "关于color有多少个字母?", "intermediate_steps": intermediate_steps }) print(f'\n轮询第{count}次',output,'过程对象：',type(output)) count +=1 # 判断代理是否有结束标记，有的话结束 if isinstance(output, AgentFinish): final_result = output.return_values["output"] break else: # 从AgentAction 中提取要执行的 tool（工具名称），跟tool_input（工具的输入参数） tool = { "get_word_length": get_word_length }[output.tool] observation = tool.run(output.tool_input) # 运行工具函数，并且得到结果 intermediate_steps.append((output, observation))

查看结果，可以看到出来了回复

6、简化代码

from langchain.chat_models import ChatOpenAI # 创建语言模型 llm = ChatOpenAI(temperature=0) # 构建工具 from langchain.agents import tool @tool def get_word_length(word: str) -> int: """返回单词的长度""" return len(word) # 创建工具清单 tools = [get_word_length] # 绑定工具 from langchain.tools.render import format_tool_to_openai_function # 格式化函数作为符合openai格式的描述体 llm_with_tools = llm.bind( functions=[format_tool_to_openai_function(t) for t in tools] ) # 创建代理 from langchain.agents.format_scratchpad import format_to_openai_functions from langchain.agents.output_parsers import OpenAIFunctionsAgentOutputParser from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个非常优秀的人，但是不会计算单词的长度"), ("user", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad"), # 消息体当中添加一个 key，记录代理的使用过程 ]) agent = { "input": lambda x: x["input"], "agent_scratchpad": lambda x: format_to_openai_functions(x['intermediate_steps']) # 代理的使用过程 } | prompt | llm_with_tools | OpenAIFunctionsAgentOutputParser() # 使用代理 from langchain.agents import AgentExecutor agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True) agent_executor.invoke({"input": "关于color有多少个字母?"})

看看结果：

关于AgentExecutor这个方法常用参数说明

max_iterations 默认迭代工具15次，可以设置
return_intermediate_steps 是否返回代理的中间步骤轨迹在最后除了最终输出之外

关于AgentExecutor的源码
https://github.com/langchain-ai/langchain/blob/804390ba4bcc306b90cb6d75b7f01a4231ab6463/libs/langchain/langchain/agents/agent.py#L917

7、加上历史对话

为了做到这一点，需要添加内存来解决这个问题：

1、在提升中添加内存变量的位置

2、跟踪聊天记录

首先在提示中添加一个内存位置。通常使用带有key的消息添加占位符来实现此目的的“chat_history”
这个放入的位置是有讲究，放到新用户输入的之上（符合对话流程）MessagesPlaceholder注意看 prompt 该位置

from langchain.agents.format_scratchpad import format_to_openai_functions from langchain.agents.output_parsers import OpenAIFunctionsAgentOutputParser from langchain.prompts import MessagesPlaceholder, ChatPromptTemplate from langchain.chat_models import ChatOpenAI from langchain.agents import tool, AgentExecutor from langchain.tools.render import format_tool_to_openai_function # 创建语言模型 llm = ChatOpenAI(temperature=0) # 构建工具 @tool def get_word_length(word: str) -> int: """返回单词的长度""" return len(word) tools = [get_word_length] # 绑定工具 llm_with_tools = llm.bind( functions=[format_tool_to_openai_function(t) for t in tools] ) MEMORY_KEY = "chat_history" prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个非常优秀的人，但是不会计算单词的长度"), MessagesPlaceholder(variable_name=MEMORY_KEY), ("user", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad"), ]) chat_history = [] agent = { "input": lambda x: x["input"], "agent_scratchpad": lambda x: format_to_openai_functions(x['intermediate_steps']), "chat_history": lambda x: x["chat_history"] } | prompt | llm_with_tools | OpenAIFunctionsAgentOutputParser() agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True) # 模拟已经对话过一次 input1 = "关于color有多少个字母?" result = agent_executor.invoke({"input": input1, "chat_history": chat_history}) # 历史消息加上聊天记录 chat_history.append(HumanMessage(content=input1)) chat_history.append(AIMessage(content=result['output'])) agent_executor.invoke({"input": "这个单词中文翻译是什么？", "chat_history": chat_history})

可以看到执行了2次

from langchain.tools import YouTubeSearchTool,Tool tool = YouTubeSearchTool() tool.run("成龙")

可以看到，能够出来2个url，此时打开网页查询一下

利用partial，格式化部分模版的参数，这里将hub模版拉下来，指定2个参数添加内容（这个模版有5个参数，这里只要2个使用）
https://github.com/langchain-ai/langchain/blob/804390ba4bcc306b90cb6d75b7f01a4231ab6463/libs/core/langchain_core/prompts/base.py#L181

如果用format_prompt 则需要指定额外参数才行，可以看源码，或者自己体验，只传2个参数会出错

from langchain.tools import YouTubeSearchTool,Tool tool = YouTubeSearchTool() # 创建工具 tools = [ Tool( name='youtube搜索', func=tool.run, description='用户如果问题有关成龙则使用这个工具' ) ] # 在prompt中绑定工具 from langchain.tools.render import render_text_description source_prompt = hub.pull("hwchase17/react-chat") prompt = source_prompt.partial( tools=render_text_description(tools),# 返回工具的文本描述 tool_names=", ".join([t.name for t in tools]), ) # 绑定工具 from langchain.llms import OpenAI llm = OpenAI(temperature=0) llm_with_stop = llm.bind(stop=["\nObservation"]) # 参考ReAct架构，添加一个停止词 # 创建代理，可以看到这里对于模版剩下3个提示词，进行了操作， from langchain.agents.output_parsers import ReActSingleInputOutputParser from langchain.agents.format_scratchpad import format_log_to_str agent = { "input": lambda x: x["input"],# 用户输入 "agent_scratchpad": lambda x: format_log_to_str(x['intermediate_steps']),# 一个是代理执行步骤 "chat_history": lambda x: x["chat_history"] # 聊天记录 } | prompt | llm_with_stop | ReActSingleInputOutputParser() # 创建代理执行执行器 from langchain.agents import AgentExecutor from langchain.memory import ConversationBufferMemory memory = ConversationBufferMemory(memory_key="chat_history") agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, memory=memory)

看看结果，2轮对话

4、AgentType

在前面，我们看到了工具创建好了，都通过自己创建Agent进行，使用，现在我们可以通过现成的封装直接使用，关于更多功能

查看源代码

准备工作，去这个平台注册一个key
https://serpapi.com/manage-api-key

安装一个包

pip install google-search-results

from langchain.agents import Tool from langchain.agents import AgentType from langchain.memory import ConversationBufferMemory from langchain.llms import OpenAI from langchain.utilities import SerpAPIWrapper from langchain.agents import initialize_agent import os os.environ["SERPAPI_API_KEY"] = "" search = SerpAPIWrapper() tools = [ Tool( name='物流工具', func=search.run, description='这对于回答相关船期与货物跟踪的时候非常有用，如果计算问题不要使用这个工具' ) ] llm = OpenAI(temperature=0) memory = ConversationBufferMemory(memory_key="chat_history") # 使用现成的代理类创建代理 agent_executor = initialize_agent( tools, llm, agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION, verbose=True, memory=memory, )

关于AgentType.CONVERSATIONAL_REACT_DESCRIPTION

执行一下看看结果

agent_executor.invoke({'盐田到洛杉矶会经过哪些港口'})

5、多工具组合以及AgentExecutorIterator使用

AgentExecutorIterator 在 langchain 中主要用于:

批量执行代理AgentExecutorIterator 可以将一组输入数据,批量传递给代理,并获取代理的批量输出。
迭代执行代理AgentExecutorIterator 可以迭代执行代理,每次传递一个输入,获取一个输出。
提供进度提示AgentExecutorIterator 可以提供进度提示,显示当前执行到第几个输入。

step1：创建模型跟链

from langchain.chains import LLMMathChain from langchain.chat_models import ChatOpenAI llm = ChatOpenAI(temperature=0, model="gpt-4") llm_math_chain = LLMMathChain.from_llm(llm=llm, verbose=True)

step2：创建2个工具

import pydantic.v1 as pydantic from langchain.agents.tools import Tool primes = {998: 7901, 999: 7907, 1000: 7919} class CalculatorInput(pydantic.BaseModel): """计算器的输入""" question: str = pydantic.Field() class PrimeInput(pydantic.BaseModel): """主输入""" n: int = pydantic.Field() def is_prime(n: int) -> bool: if n <= 1 or (n % 2 == 0 and n > 2): return False for i in range(3, int(n ** 0.5) + 1, 2): if n % i == 0: return False return True def get_prime(n: int, primes: dict = primes) -> str: """计算素数""" return str(primes.get(int(n))) async def aget_prime(n: int, primes: dict = primes) -> str: """代理""" return str(primes.get(int(n))) # 工具清单 tools = [ Tool( name="GetPrime", func=get_prime, description="返回第 n 个素数的工具", args_schema=PrimeInput, # 以pydantic 进行约束字段 coroutine=aget_prime, # 异步 ), Tool.from_function( func=llm_math_chain.run, name="Calculator", description="当您需要计算数学表达式时很有用", args_schema=CalculatorInput, # 以pydantic 进行约束字段 coroutine=llm_math_chain.arun, # 异步 ), ]

step3：创建代理并且使用代理

# 创建代理 agent = initialize_agent( tools, llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION, verbose=True ) # 使用代理 question = "第 998 个、第 999 个和第 1000 个素数的乘积是多少？" for step in agent.iter(question): if output := step.get("intermediate_step"): action, value = output[0] if action.tool == "GetPrime": print(f"正在检查 {value} 是否为质数...") assert is_prime(int(value)) # Ask user if they want to continue _continue = input("代理是否应该继续 (Y/n)?:\n") if _continue != "Y": break

查看结果，很有意思

1、检查用户输入的998，999，1000 通过工具is_prime检查是否为素数

2、然后最后得到3个素数，调用get_prime工具进行计算

3、整个过程当中因为使用了ReAct ，能够看到关键词
- Action
- Action input
- Observation
- Thought
- Answer
- Final Answer

4、什么时候使用另外一个工具，也做到了，这就是prompt工程的魅力

平替方法：

from langchain.agents import AgentExecutorIterator AgentExecutorIterator( agent_executor=agent_executor, inputs=[{'input': "第 998 个、第 999 个和第 1000 个素数的乘积是多少？"}] ) # 迭代执行代理,获取输出 outputs = [] for output in iterator: outputs.append(output)