文章总结: 本文介绍Langchain代理(agent)机制,结合ReAct框架实现LLM多步推理与工具调用。演示了计算器工具(LLMMathChain)如何将自然语言转为表达式执行,以及维基百科API的知识查询功能。针对网络限制提出替代方案如自定义Bing搜索工具,并分析多轮思考的token消耗与防频率限制策略。关键点包括代理结构、思维链(CoT)应用及工具注册方法。
综合评分: 85
文章分类: AI安全,安全工具,安全开发,技术标准,安全运营
Langchain6_代理agent
原创
羽泪云小栈
羽泪云小栈
羽泪云小栈
2026年4月9日 10:00
陕西
代理agent
不同于网络代理的proxy,这里的代理指代理人,agent。
提供文本或其它来源,llm从互联网上学习新知识进行回答或推理
在这里,agent=llm+工具调用
Langchian_代理+ReAct框架(llm(思维链)+循环执行+工具调用)+tool注册+docker安装部署(python调用docker)
前置
安装维基百科、其它库,但是维基百科访问不了,干脆让llm写一个请求bing的就行
pip install -U wikipedia
LLMMathChain requires the numexpr package. Please install it with pip install numexpr.
tools = load_tools(
["llm-math","wikipedia"],
llm=llm
)
调用计算器llm-math
这个计算器实现方式是,通过llm生成python代码,再执行的,所以这就是为什么它叫llm-math
return Tool(
name="Calculator",
description="Useful for when you need to answer questions about math.",
func=LLMMathChain.from_llm(llm=llm).run,
coroutine=LLMMathChain.from_llm(llm=llm).arun,
)
# coding: utf-8
import os
from dotenv import load_dotenv, find_dotenv
from langchain_openai import ChatOpenAI
from langchain.agents import load_tools, initialize_agent
from langchain.agents import AgentType
from langchain_core.tracers import ConsoleCallbackHandler
load_dotenv(find_dotenv())
llm = ChatOpenAI(
# This is the default and can be omitted
api_key=os.environ.get("KEY2"),
base_url=os.environ.get("base_url2"),
temperature=0.0,#让预测不让那么随机,偏平稳
model=os.environ.get("model2"),
verbose=True
)
tools = load_tools(
["llm-math","wikipedia"],
llm=llm
)
agent= initialize_agent(
tools, #第二步加载的工具
llm, #第一步初始化的模型
agent=AgentType.CHAT_ZERO_SHOT_REACT_DESCRIPTION, #代理类型
handle_parsing_errors=True, #处理解析错误
verbose = True #输出中间步骤
)
agent("1乘以2,再加5的结果,乘以50,再加上1776,再减去2010的结果?")
这是老版本内容,新版本的话无非就是自己要构造下prompt,先看看老版的prompt
print(agent.agent.llm_chain.prompt.messages[0].prompt.template)
"""
Answer the following questions as best you can. You have access to the following tools:
Calculator: Useful for when you need to answer questions about math.
wikipedia: A wrapper around Wikipedia. Useful for when you need to answer general questions about people, places, companies, facts, historical events, or other subjects. Input
should be a search query.
The way you use the tools is by specifying a json blob.
Specifically, this json should have a `action` key (with the name of the tool to use) and a `action_input` key (with the input to the tool going here).
The only values that should be in the "action" field are: Calculator, wikipedia
The $JSON_BLOB should only contain a SINGLE action, do NOT return a list of multiple actions. Here is an example of a valid $JSON_BLOB:
{{
“action”: $TOOLNAME,
“actioninput”: $INPUT
}}
ALWAYS use the following format:
Question: the input question you must answer
Thought: you should always think about what to do
Action:
$JSON_BLOB
Observation: the result of the action
... (this Thought/Action/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question
Begin! Reminder to always use the exact characters `Final Answer` when responding.
这个prompt直接用的话,会提示:
raise ValueError(f"Prompt missing required variables: {missing_vars}")
ValueError: Prompt missing required variables: {'tool_names', 'agent_scratchpad', 'tools'}
干脆自己生成一个吧,为了显示过程,需要引入AgentExecutor
from langchain.agents import create_react_agent,AgentType,AgentExecutor
from langchain_community.agent_toolkits.load_tools import load_tools
...
prompt = """你是一个助手,可以使用以下工具:
{tools}
工具名称:{tool_names}
你必须按以下格式回答:
Question: 用户的问题
Thought: 思考应该做什么
Action: 工具名称,必须是 [{tool_names}]
Action Input: 工具的输入
Observation: 工具返回的结果
... (重复 Thought/Action/Action Input/Observation)
Thought: 我知道答案了
Final Answer: 最终答案
开始!
Question: {input}
Thought: {agent_scratchpad}"""
prompt=PromptTemplate.from_template(prompt)
agent=create_react_agent(
llm=llm,
tools=tools,
prompt=prompt,
)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
handle_parsing_errors=True
)
# 执行
result = agent_executor.invoke({"input": "1乘以2,再加5的结果,乘以50,再加上1776,再减去2010的结果?"})
重点1.让llm自己生成代码自己跑? 按理说它只有文本能力
看 LLMMathChain
中间有个正则,还有_evalate_expression(expression)这个方法
所以它实际做了一件事,让llm将用户的问题,转化成表达式,然后交给python去提取这个表达式,并真正地计算。
重点2.这两次Thought很有意思:
第一次让它规划,分析问题,用什么工具
第二次是分析结果,决定下一步
平时使用大模型平台的时候,都会看到诸如think或者深度思考的方式,原来就是这样通过让llm强制输出思考过程。
这就像是思维链的本质(Chain of Thought )CoT,让LLM把思考过程写出来,可以检索一下它的想法,同时提高答案质量。
这和之前的路由链是不是有些类似,路由链只做一次分类判断和动作
但这个可以多轮思考,多步推理,其实也…更贵吧,因为导致token变多hhh
回归正题,基于重点1的操作来看,那么这两次Thought,绝不是只调用了一次llm,因为重点1只是提取表达式,并计算。
但第二步有一个Thought怎么来的呢
先看老版本的实现方式:
(function) def initialize_agent(
tools: Sequence[BaseTool],
llm: BaseLanguageModel,
agent: AgentType | None = None,
callback_manager: BaseCallbackManager | None = None,
agent_path: str | None = None,
agent_kwargs: dict | None = None,
*,
tags: Sequence[str] | None = None,
**kwargs: Any
) -> AgentExecutor
返回值类型是这个:AgentExecutor
这个py文件呢,看着pan方法:
反正它说了基于过去的历史和现在的输入…
_construct_scratchpad
这是在拼接历史记录…
具体的我也看不懂了,总之按我的理解就是,输出了几次Thought,就是调用了几次llm。
重点3:至于为什么没有触发429这种频率限制:
对于之前我的for循环中的invoke来讲,调用间隔几乎是无延迟的,所以立马被429
而由于这次的两个Thought之间,第一次调用完是要回到python中去执行计算表达式给出结果的,估计这个过程占了那么几秒,导致没有触发429
补充:
可以用全局调试信息打印看看
from langchain.globals import set_debug, set_verbose
set_debug(True)
set_verbose(True)
维基百科api
它其实是相当于一个爬虫api,联网获取知识
...
tools = load_tools(
["llm-math","wikipedia"],
llm=llm
)
...
input2="保尔.柯察金是一个怎样的人?"
result=agent_executor.invoke({"input":input2})
第一次Thought正常,第二次访问超时,需要用proxy
这个就演示到这里,要么proxy、要么…国内站(可能注册给api,可能需要一点额外措施),
也可以让llm写个简单的requests请求,写到函数里,之后load一下,但需要说明的是,比如通过bing去搜索关键词,是可以得到一些标题和摘要的,但是毕竟每个链接不同,点进去后的网站架构都不同,不好提取正文,这就是为什么有个百科的重要性,内容架构是固定的。
这里就用简单点的方法演示下就行了,由于不是langchain自带的,需要包装一波,”注册”一下
def bing_search(query: str, max_results: int = 10) -> str:
...
return results
...
from bingapi import bing_search
from langchain_core.tools import Tool
...
bing_tool=Tool(
name="bingsearch",
func=bing_search,
description="用于搜索中文知识、百科内容,替代维基百科API"
)
tools1 = load_tools(
["llm-math"],
llm=llm
)
tools1=tools1+[bing_tool]
agent1=create_react_agent(
llm=llm,
tools=tools1,
prompt=prompt,
)
agent_executor1 = AgentExecutor(
agent=agent1,
tools=tools1,
verbose=True,
handle_parsing_errors=True
)
input2="怎么理解除去巫山不是云?"
result=agent_executor1.invoke({"input":input2})
#wikipedia.summary()
现在的问题是… 似乎没有传到值…导致搜索结果错误?
但是函数的接受参数也是query,没问题啊…
但是惊喜还在后面:
不愧是ReAct:
首先是prompt,明确它的输出思维过程,并且自我纠正的思维格式
其次是AgentExecutor 做循环,能够让llm看到完整的对话历史(包括之前的错误),llm在发现问题,重新决策
最终达到自动纠错、自动重搜的目的
最终结果:
自定义工具1-代码执行
本来是PythonREPLTool的,倒是好像被弃了
REPL 是”Read–Eval–Print Loop”(读取-求值-打印-循环)的缩写,它是一种简单的、交互式的编程环境。在REPL环境中,用户可以输入一条或多条编程语句,系统会立即执行这些语句并输出结果。可以逐行输入Python代码,每输入一行代码并回车,就会立即执行这行代码并打印输出结果
和之前的 LLMMathChain 生成表达式一样,这里是让llm生成python代码,交给Python 解释器执行
https://docs.langchain.com/oss/python/integrations/tools/python
好像因为安全性的问题,在新版本已经删除,且移到langchain_experimental了
要么docker,要么云沙箱
好像是这么个道理,llm生成的python代码,你敢不敢本地跑?
emmmm,我还是安装个docker吧…
可以写个函数了
import subprocess
import tempfile
def docker_python_execute(code: str) -> str:
"""在 Docker 容器中安全执行 Python 代码。用于计算、数据处理等场景。"""
# 写入临时文件(避免命令行转义问题)
with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False, encoding='utf-8') as f:
f.write(code)
temp_path = f.name
try:
# 运行 Docker 容器
# --rm: 自动删除容器
# --network none: 禁止网络访问
# --memory 256m: 内存限制
# --read-only: 只读根文件系统(需要临时目录可写,这里用 /tmp)
# -v 挂载临时文件
result = subprocess.run(
[
"docker", "run", "--rm",
"--network", "none",
"--memory", "256m",
"--read-only",
"-v", f"{temp_path}:/script.py:ro",
"python:3.11-slim",
"python", "/script.py"
],
capture_output=True,
text=True,
timeout=10
)
output = result.stdout
if result.stderr:
output += f"\n[stderr]: {result.stderr}"
return output if output.strip() else "代码执行成功(无输出)"
except subprocess.TimeoutExpired:
return "执行超时(10秒)"
except Exception as e:
return f"执行失败: {e}"
finally:
# 清理临时文件
if os.path.exists(temp_path):
os.unlink(temp_path)
print(docker_python_execute("print(7+3)"))
#输出了10
代码
# coding: utf-8
import os
import langchain
langchain.verbose = True
from dotenv import load_dotenv, find_dotenv
from langchain_openai import ChatOpenAI
#from langchain.agents import load_tools, initialize_agent
from langchain.agents import create_react_agent,AgentType,AgentExecutor
from langchain_community.agent_toolkits.load_tools import load_tools
from langchain_community.tools import tool
from langchain_core.tracers import ConsoleCallbackHandler
from langchain_core.prompts import ChatPromptTemplate,PromptTemplate
import subprocess
import tempfile
from langchain_core.tracers import ConsoleCallbackHandler
load_dotenv(find_dotenv())
llm = ChatOpenAI(
# This is the default and can be omitted
api_key=os.environ.get("KEY2"),
base_url=os.environ.get("base_url2"),
temperature=0.0,#让预测不让那么随机,偏平稳
model=os.environ.get("model2"),
verbose=True
)
@tool
def docker_python_execute(code: str) -> str:
"""在 Docker 容器中安全执行 Python 代码。用于计算、数据处理等场景。"""
# 写入临时文件(避免命令行转义问题)
with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False, encoding='utf-8') as f:
f.write(code)
temp_path = f.name
try:
# 运行 Docker 容器
# --rm: 自动删除容器
# --network none: 禁止网络访问
# --memory 256m: 内存限制
# --read-only: 只读根文件系统(需要临时目录可写,这里用 /tmp)
# -v 挂载临时文件
result = subprocess.run(
[
"docker", "run", "--rm",
"--network", "none",
"--memory", "256m",
"--read-only",
"-v", f"{temp_path}:/script.py:ro",
"python:3.11-slim",
"python", "/script.py"
],
capture_output=True,
text=True,
timeout=10
)
output = result.stdout
if result.stderr:
output += f"\n[stderr]: {result.stderr}"
return output if output.strip() else "代码执行成功(无输出)"
except subprocess.TimeoutExpired:
return "执行超时(10秒)"
except Exception as e:
return f"执行失败: {e}"
finally:
# 清理临时文件
if os.path.exists(temp_path):
os.unlink(temp_path)
tools1 = load_tools(
["llm-math"],
llm=llm
)
tools1=tools1+[docker_python_execute]
prompt=prompt = """你是一个助手,可以使用以下工具:
{tools}
工具名称:{tool_names}
你必须按以下格式回答:
Question: 用户的问题
Thought: 思考应该做什么
Action: 工具名称,必须是 [{tool_names}]
Action Input: 工具的输入
Observation: 工具返回的结果
... (重复 Thought/Action/Action Input/Observation)
Thought: 我知道答案了
Final Answer: 最终答案
开始!
Question: {input}
Thought: {agent_scratchpad}"""
prompt=PromptTemplate.from_template(prompt)
agent1=create_react_agent(
llm=llm,
tools=tools1,
prompt=prompt,
)
agent_executor1 = AgentExecutor(
agent=agent1,
tools=tools1,
verbose=True,
handle_parsing_errors=True,
callbacks=[ConsoleCallbackHandler()]
)
customer_list = [["小李",100],
["小吴",45],
["小刘",74],
["小七",58],
["小九",60]]
input1=f"""假设这里是5人的名字和对应的成绩,请按成绩从小到大进行排序:{customer_list}"""
agent_executor1.invoke({"input":input1})
如果,把代码中那个 “–rm” 取消了的话,不让它自动删容器,那边是可以看到docker确实是在执行的
result = subprocess.run(
[
"docker", "run",
...
input1=f"""假设这里是5人的名字和对应的成绩,请按成绩从大到小进行排序:{customer_list}"""
agent_executor1.invoke({"input":input1})
自定义工具2-情绪助手?
这里展示两个用法:
1.一是@tool将函数注册进langchain识别中的工具中,且函数的”””部分,会作为描述被llm读取
2.二是全局调试信息的用法
或者langchain_core.globals
from langchain.globals import set_debug, set_verbose
set_debug(True)
set_verbose(True)
代码
...
from langchain.globals import set_debug, set_verbose
set_debug(True)
set_verbose(True)
...
@tool
def wishuhappy(text:str)->str:
"""当用户表达了不开心的原话时,用这个工具安慰用户"""
return "yly允许你哭一会儿,希望你能重拾心情照顾好自己。"
agent1=create_react_agent(
llm=llm,
tools=[wishuhappy],
prompt=prompt,
)
agent_executor1 = AgentExecutor(
agent=agent1,
tools=[wishuhappy],
verbose=True,
handle_parsing_errors=True,
callbacks=[ConsoleCallbackHandler()]
)
input1="""今天我心情不好。"""
agent_executor1.invoke({"input":input1})
可以看到这个过程里,函数是成功执行了的。
总结一下
agent充分利用了llm的文本能力+工具的能力(计算、搜索等…)
ReAct利用了llm进行自我纠错的这种思维链+循环方式:
实现:prompt+create_react_agent+AgentExecutor
自定义langchain工具 @tool的使用
思维链:Thought → Action → Observation → (重复) → Final Answer
注意
- 1. ChatPromptTemplate,PromptTemplate 前者是角色分区与多轮对话(字典形式),后者是纯文本(字符串)
- 2. 维基访问超时,可以试试魔法
HTTPConnectionPool(host='en.wikipedia.org', port=80): Max retries exceeded with url: /w/api.php?list=search&srprop=&srlimit=3&limit=3&srsearch=%E4%BF%9D%E5%B0%94%C2%B7%E6%9F%AF%E5%AF%9F%E9%87%91&format=json&action=query (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x000002834F315A00>: Failed to establish a new connection: [WinError 10060] 由于连接方在一段时
间后没有正确答复或连接的主机没有反应,连接尝试失败。'))
3.docker-desktop-windows安装
https://www.docker.com/products/docker-desktop/ 下载windows版
https://www.jianshu.com/p/7b47c0b79f20
(如果想指定位置安装比如D盘):
win+x-> powershell管理员模式
Start-Process -Wait -FilePath ".\Docker Desktop Installer.exe" -ArgumentList "install -accept-license --installation-dir=D:\Docker --wsl-default-data-root=D:\Docker\WSL --windows-containers-default-data-root=D:\\Docker\\WindowsContainers"
使用:
docker --version 安装成功了就有显示的
wsl --update #提示我更新
管理员身份运行docker desktop,否则有些权限问题
# 停止所有 Docker 相关进程
#ps 管理员下
Get-Process | Where-Object {$_.ProcessName -like "*docker*"} | Stop-Process -Force
Stop-Service com.docker.service -Force
命令行安装docker-python
docker pull python:3.11-slim #挂代理先
docker run --rm python:3.11-slim python -c "print(2+3)" #回复5就是成功
4.把一个普通函数注册成 LangChain 能识别的工具:
之间在函数前用@tool装饰器,这属于新版推荐的方法
另一种就算维基百科api那一节用的,Tool()实例化也可以
免责声明:
本文所载程序、技术方法仅面向合法合规的安全研究与教学场景,旨在提升网络安全防护能力,具有明确的技术研究属性。
任何单位或个人未经授权,将本文内容用于攻击、破坏等非法用途的,由此引发的全部法律责任、民事赔偿及连带责任,均由行为人独立承担,本站不承担任何连带责任。
本站内容均为技术交流与知识分享目的发布,若存在版权侵权或其他异议,请通过邮件联系处理,具体联系方式可点击页面上方的联系我。
本文转载自:羽泪云小栈 羽泪云小栈
羽泪云小栈《Langchain6_代理agent》