SEATA分布式事务——AT模式速
简介
AI Agent 不仅仅是一个能聊天的机器人（如普通的 ChatGPT），而是一个能够感知环境、进行推理、自主决策并调用工具来完成特定任务的智能系统，更够完成更为复杂的AI场景需求。
AI Agent
功能
根据查阅的资料，agent的功能点如下：
Agent = LLM + 规划 + 记忆 + 工具使用
LLM： 用于回答，推理的AI模型
记忆： 短期记忆（对话历史），长期记忆（RAG知识库）
规划： 任务的执行流
工具： Agent可以调用的外部函数
一个简单的Agent
案例
案例描述：agent可以使用两个工具：1. 记录公司数据的RAG知识库 2. python精确计算器
import os
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.messages import ToolMessage
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter
os.environ["DASHSCOPE_API_KEY"] = "api_key"
@tool
def calculator(expression: str) -> str:
"""
计算数学表达式。需要精确计算时使用。
参数:
expression: 数学算式，如 "2 + 2" 或 "500 * 0.8"。
返回:
str: 计算结果，如 "4.0" 或 "400.0"。
"""
print(f" [??? 工具调用] 计算器正在计算: {expression}")
try:
return str(eval(expression))
except Exception as e:
return f"计算错误: {e}"
@tool
def rag_search(query: str) -> str:
"""
从数据库中搜索与查询公司内部相关的文档，包括公司计划名，代号，截止日期等详细信息。
参数:
query (str): 要搜索的查询字符串。
返回:
str: 与查询相关的文档内容。
"""
# 从RAG数据库中检索文档
raw_text = """
【公司内部机密：代号“深蓝计划”】
1. 项目目标：开发一款能听懂猫语的翻译器。
2. 核心技术：基于Transformer的“喵声波”分析算法。
3. 团队暗号：如果有人问“今天天气怎么样？”，必须回答“我想吃鱼”。
4. 截止日期：2026年12月31日。
5. 经费预算：仅剩50元人民币，主要用于购买猫条。
"""
RAG_PATH = "faiss_index"
docs = [Document(page_content=raw_text)]
text_splitter = RecursiveCharacterTextSplitter(chunk_size=25, chunk_overlap=5)
split_docs = text_splitter.split_documents(docs)
embeddings = DashScopeEmbeddings(model="text-embedding-v1")
if os.path.exists(RAG_PATH):
print("公司内部数据库已存在")
ragdb = FAISS.load_local(RAG_PATH, embeddings, allow_dangerous_deserialization=True)
else:
print("创建公司内部数据库")
ragdb = FAISS.from_documents(split_docs, embeddings)
ragdb.save_local(RAG_PATH)
return "\n\n".join(doc.page_content for doc in ragdb.similarity_search(query, k=2))
def run_agent(query:str):
#初始化模型
tool_maps={
"rag_search": rag_search,
"calculator": calculator
}
llm = ChatTongyi(model_name="qwen-plus")
tool_llm = llm.bind_tools(tools=list(tool_maps.values()))
message = [HumanMessage(content=query)]
for i in range(5):
print("="*20+"\n第"+str(i+1)+"轮\n"+query+"\n"+"="*20)
response = tool_llm.invoke(message)
message.append(response)
print(f"需要调用{len(response.tool_calls)}个方法")
if not response.tool_calls:
print("最终结果：" + response.content)
return
for tool_call in response.tool_calls:
call_id = tool_call["id"]
func_name = tool_call["name"]
func_args = tool_call["args"]
# 安全检查：确保模型调用的工具真的存在
if func_name in tool_maps:
# 运行 Python 函数
tool_func = tool_maps[func_name]
tool_output = tool_func.invoke(func_args)
print("工具调用：" + func_name + "，参数：" + str(func_args) + "，结果：" + tool_output)
else:
tool_output = f"错误: 工具 {func_name} 不存在。"
message.append(
ToolMessage(
content=tool_output,
tool_call_id=call_id,
name=func_name,
)
)
if __name__ == "__main__":
run_agent("公司计划是什么")
run_agent("公司的经费预算是多少，如果预算预算提高46%后多少")
run_agent("今天天气真好")
代码解析
要实现复杂的工具调用，必须实现AI的多轮对话，在langchain框架中，提供了大量的prompt模板，让开发者不需要过度想一些基础的prompt实现。
上面代码的执行流程如下：
初始化2个工具函数->绑定LLM与工具->通过循环进行多轮对话
初始化2个工具函数
这里的rag_search上一篇文章讲了具体实现，这里就不废话了。
@tool
def calculator(expression: str) -> str:
"""
计算数学表达式。需要精确计算时使用。
参数:
expression: 数学算式，如 "2 + 2" 或 "500 * 0.8"。
返回:
str: 计算结果，如 "4.0" 或 "400.0"。
"""
print(f" [??? 工具调用] 计算器正在计算: {expression}")
try:
return str(eval(expression))
except Exception as e:
return f"计算错误: {e}"
@tool
def rag_search(query: str) -> str:
......
工具函数的格式，主要有3个方面：
工具修饰： 利用@tool修饰器修饰
函数的描述： 这里放函数的描述，大模型通过这个描述定位工具，因此这部分必须详细，可以参考上面:
函数的描述
函数的参数+例子
函数的返回+例子
工具的实现： 返回值要是字符串
如下：
@tool
def func_name(arg) -> str:
"""
描述
"""
......
绑定LLM与工具
工具的绑定非常的简单，只需要简单的bind_tools类方法就行
tool_maps={
"rag_search": rag_search,
"calculator": calculator
}
llm = ChatTongyi(model_name="qwen-plus")
tool_llm = llm.bind_tools(tools=list(tool_maps.values()))
通过循环进行多轮对话（重点）
工具的调用流程：
提示词->LLM->要调用的工具->LLM->结果
message = [HumanMessage(content=query)]
for i in range(5):
print("="*20+"\n第"+str(i+1)+"轮\n"+query+"\n"+"="*20)
response = tool_llm.invoke(message)
message.append(response)
print(f"需要调用{len(response.tool_calls)}个方法")
if not response.tool_calls:
print("最终结果：" + response.content)
return
for tool_call in response.tool_calls:
call_id = tool_call["id"]
func_name = tool_call["name"]
func_args = tool_call["args"]
# 安全检查：确保模型调用的工具真的存在
if func_name in tool_maps:
# 运行 Python 函数
tool_func = tool_maps[func_name]
tool_output = tool_func.invoke(func_args)
print("工具调用：" + func_name + "，参数：" + str(func_args) + "，结果：" + tool_output)
else:
tool_output = f"错误: 工具 {func_name} 不存在。"
message.append(
ToolMessage(
content=tool_output,
tool_call_id=call_id,
name=func_name,
)
)
在调用bind_tools方法后，大模型的返回对象会多出tool_calls字段的数组数据，用于存放需要调用工具的参数，函数名，在调用函数后，将调用函数的结果封装成ToolMessage传入，再继续调用大模型。
注意：在调用LLM时可能LLM会不断要求Tool，由此可能发生死循环，因此要限制循环次数。
安全与审思
风险评估
近些年，ai提示词注入频频发生，根据上面的案例：
@tool
def calculator(expression: str) -> str:
"""
计算数学表达式。需要精确计算时使用。
参数:
expression: 数学算式，如 "2 + 2" 或 "500 * 0.8"。
返回:
str: 计算结果，如 "4.0" 或 "400.0"。
"""
print(f" [??? 工具调用] 计算器正在计算: {expression}")
try:
return str(eval(expression))
except Exception as e:
return f"计算错误: {e}"
LLM在调用这个工具时使用了eval，这就造成了风险注入点，不法分子可能利用这个漏洞，获取电脑权限。
修复思路
修复上面漏洞，可以参考以下几个思路：
通过指令提示词，让LLM忽略，并终止危险代码调用该函数
在函数中用正则匹配危险代码，或设置白名单
替换eval函数，将其换为更加安全的方法，如手动写死运算势俣颓佣