主要介绍 AI agent 相关的学习，参考项目 wow-agent

# 什么是 AI Agent ?

LLMs 强大的语言理解和生成能力，为构建 Agents 奠定了基础。然而，LLMs 的局限性也显而易见：

知识局限性：LLMs 的知识仅限于训练数据，无法获取实时信息和外部知识库。
行动局限性：LLMs 无法与外界交互，无法执行实际操作。为了克服这些局限性，谷歌的研究人员在《New whitepaper Agents》详细论述了 “Agent” 的概念，将 LLMs 与工具和编排层相结合，赋予其自主行动的能力。

AI Agent 是目前大模型应用领域的主流研究方向，其核心价值在于借助先进 AI 技术，打造出更加智能、更加自主的系统。

Agent 不仅在处理复杂自然语言任务方面游刃有余，更在图像生成、数据分析、教育指导等多个前沿领域展现出卓越的应用潜力。

关于 AI Agent 框架的定义，目前最广为流传的版本出自前 OpenAI 应用研究主管翁丽莲 (Lilian Weng) 的一篇 blog: LLM Powered Autonomous Agents

在博客里，Lilian 将 Agents 定义为 LLM + memory + planning skills + tool use，即大语言模型、记忆、任务规划、工具使用的集合。

其中，LLM 是 Agent 的大脑，属于 “中枢” 模型，要求有以下 3 种能力：

planning skills：对问题进行拆解得到解决路径，既进行任务规划
tool use：评估自己所需的工具，进行工具选择，并生成调用工具请求
memory：短期记忆包括工具的返回值，已经完成的推理路径；长期记忆包括可访问的外部长期存储，例如知识库最早由不仅具备了感知、记忆和决策的全面能力，更能在复杂多变的环境中自主执行任务，并通过与外部工具的紧密协作，显著提升任务处理的效率与准确性。

更多相关知识可见大模型关键技术与应用，这篇文章介绍了自 ChatGPT 发布以来，大模型关键技术和应用的主要进展。

# Agents 的核心组件

Agents 的核心组件
一个完整的 Agent 主要由三个核心组件构成：

1. 模型 (Model)

角色：作为 Agent 的 “大脑”，负责理解用户输入，进行推理和规划，并选择合适的工具进行执行。
类型：常用的模型包括 ReAct、Chain-of-Thought、Tree-of-Thought 等，它们提供不同的推理框架，帮助 Agent 进行多轮交互和决策。
重要性：模型是 Agent 的核心，其推理能力决定了 Agent 的行动效率和准确性。

2. 工具 (Tools)

角色：作为 Agent 与外界交互的 “桥梁”，允许 Agent 访问外部数据和服务，执行各种任务。
类型：工具可以是各种 API，例如数据库查询、搜索引擎、代码执行器、邮件发送器等。
重要性：工具扩展了 Agent 的能力，使其能够执行更复杂的任务。

3. 编排层 (Orchestration Layer)

角色：负责管理 Agent 的内部状态，协调模型和工具的使用，并根据目标指导 Agent 的行动。
类型：编排层可以使用各种推理框架，例如 ReAct、Chain-of-Thought 等，帮助 Agent 进行规划和决策。
重要性：编排层是 Agent 的 “指挥中心”，负责协调各个组件，确保 Agent 的行动符合目标。

# Agent 的运作机制

从输入到输出，Agent 的运作过程可以概括为以下几个步骤：

**1. 接收输入 **：Agent 接收用户的指令或问题。

**2. 理解输入 **：模型理解用户的意图，并提取关键信息。

**3. 推理规划 **：模型根据用户输入和当前状态，进行推理和规划，确定下一步行动。

**4. 选择工具 **：模型根据目标选择合适的工具。

**5. 执行行动 **：Agent 使用工具执行行动，例如查询数据库、发送邮件等。

**6. 获取结果 **：Agent 获取工具执行的结果。

**7. 输出结果 **：Agent 将结果输出给用户，或进行下一步行动。

# Agents 的优势

与传统的 LLMs 相比，Agents 具有以下优势：

知识扩展：通过工具，Agent 可以访问实时信息和外部知识库，突破训练数据的限制，提供更准确和可靠的信息。
自主行动：Agent 可以根据目标进行自主决策和行动，无需人工干预，提高效率和灵活性。
多轮交互：Agent 可以管理对话历史和上下文，进行多轮交互，提供更自然和流畅的用户体验。
可扩展性：Agent 可以通过添加新的工具和模型，扩展其功能和应用范围。

# Agents 的应用

Agents 的应用范围非常广泛，例如：

智能客服：Agent 可以自动回答用户问题，处理订单，解决客户问题，提高客户满意度。
个性化推荐：Agent 可以根据用户的兴趣和行为，推荐商品、内容、服务等，提升用户体验。
虚拟助手：Agent 可以帮助用户管理日程、预订行程、发送邮件等，提高工作效率。
代码生成：Agent 可以根据用户的需求，自动生成代码，提高开发效率。
智能创作：Agent 可以根据用户的需求，创作诗歌、小说、剧本等，激发创作灵感。
知识图谱构建：Agent 可以从文本中提取知识，构建知识图谱，用于知识管理和推理。

# Agents 的开发工具

为了方便开发 Agents，Google 提供了多种工具和平台，例如：

LangChain：一个开源库，可以帮助开发者构建和部署 Agents。LangChain 提供了一套 API，方便开发者将 LLMs 与工具和编排层结合，构建功能强大的 Agents。
LangGraph：一个开源库，可以帮助开发者构建和可视化 Agents。LangGraph 提供了一套图形化界面，方便开发者设计和测试 Agents。
Vertex AI：一个云平台，提供各种 AI 工具和服务，例如 Vertex Agent Builder、Vertex Extensions、Vertex Function Calling 等，可以帮助开发者快速构建和部署 Agents。Vertex AI 提供了强大的基础设施和工具，方便开发者进行 Agent 开发、测试、部署和管理。

# 简易 Agent 实现

在了解完相关概念后，我们正式开始学习 agent，本次使用 wow-agent 作为示例讲解 agent 实现。

通过观察市场上比较流行的开源多智能体框架，例如 metagpt、crewai、camel-ai、autogen，我们会发现这些框架安装起来有很多依赖，但是如果我们只是想要实现一个简易的功能，就不需要这么多的依赖环境，所以，wow-agent 应运而生。

wow-agent 致力于在代码行数和依赖库数量之间取得均衡的最小值，用最划算的方式在本地搭建 AI Agent，嵌入到生产工作环节中。

# 环境准备

先准备需要的 python 库： OpenAI、Python-dotenv
pip install openai python-dotenv

OpenAI 库是 OpenAI 官方提供的 Python SDK，旨在帮助开发者轻松调用 OpenAI 的 API，实现各种 AI 功能。通过这个库，开发者可以快速集成 GPT、DALL・E 等先进模型，构建智能应用。
Python-dotenv 是一个用于管理环境变量的 Python 库。它可以从.env 文件中读取键值对，并将它们设置为环境变量。这个库特别适用于管理敏感信息，如 API 密钥、数据库连接字符串等。

为了调用大模型，我们需要准备：

api_key，这个需要到各家的开放平台上去申请。
base_url，这个需要到各家的开放平台上去拷贝。
所用模型名称。

在项目的根目录新建一个 txt 文件，把文件名改成.env 并在文件中写入自己的 api-key
DeepSeek_API_KEY=sk-**********************

把 DeepSeek_API_KEY 写到.env 文件的原因是为了保密，同时可以方便地在不同的代码中读取。

在大模型厂商官网注册后，能够获取到一个 api_key，一般官网文档会提供 api 结构的调用方式

对于本地 ollama 的 LLM，ollama 默认没有 api_key，需要修改请求格式，或者使用 one-api 整合使模型能够使用 openai 的调用方式，方法请自行百度，在此不做赘述。

本次演示使用第三方平台 (硅基流动) 中的 deepseek-ai/DeepSeek-V3 模型，调用参考官方 api 文档

现在，我们先把所需的环境写好

	import os
	from dotenv import load_dotenv

	# 加载环境变量
	load_dotenv ()
	# 从环境变量中读取 api_key
	api_key = os.getenv ('DeepSeek_API_KEY')
	base_url = "https://api.siliconflow.cn/v1/"
	chat_model = "deepseek-ai/DeepSeek-V3"

.env 文件需要放在项目根目录下，使用 print (os.path.abspath ('.env')) 查看.env 文件路径

构造 client

	from openai import OpenAI
	client = OpenAI (
	api_key = api_key,
	base_url = base_url
	)

有了这个 client，我们就可以去实现各种能力了

	def get_completion (prompt):
	response = client.chat.completions.create (
	model=chat_model, # 模型名称
	messages=[
	{"role": "user", "content": prompt},
	],
	)
	return response.choices [0].message.content

现在，我们先测试一下 LLM 是否可用，如果一切顺利，控制台将会 LLM 的回复

	import os
	from dotenv import load_dotenv
	from openai import OpenAI

	# 加载环境变量
	load_dotenv ()
	# 从环境变量中读取 api_key
	api_key = os.getenv ('DeepSeek_API_KEY')
	base_url = "https://api.siliconflow.cn/v1/"
	chat_model = "deepseek-ai/DeepSeek-V3"


	client = OpenAI (
	api_key = api_key,
	base_url = base_url
	)

	def get_completion (prompt):
	response = client.chat.completions.create (
	model=chat_model, # 模型名称
	messages=[
	{"role": "user", "content": prompt},
	],
	)
	return response.choices [0].message.content

	response = get_completion ("你是谁？")
	print (response)

智能体需要大量的 prompt 工程。代码是调用机器能力的工具， Prompt 是调用大模型能力的工具。Prompt 越来越像新时代的编程语言。这里使用著名的 LangGPT

# 定义 prompt

根据 LangGPT 框架，为智能体定义一些 prompt

	sys_prompt = """ 你是一个聪明的客服。您将能够根据用户的问题将不同的任务分配给不同的人。您有以下业务线：
	1. 用户注册。如果用户想要执行这样的操作，您应该发送一个带有 "registered workers" 的特殊令牌。并告诉用户您正在调用它。
	2. 用户数据查询。如果用户想要执行这样的操作，您应该发送一个带有 "query workers" 的特殊令牌。并告诉用户您正在调用它。
	3. 删除用户数据。如果用户想执行这种类型的操作，您应该发送一个带有 "delete workers" 的特殊令牌。并告诉用户您正在调用它。
	"""
	registered_prompt = """
	您的任务是根据用户信息存储数据。您需要从用户那里获得以下信息：
	1. 用户名、性别、年龄
	2. 用户设置的密码
	3. 用户的电子邮件地址
	如果用户没有提供此信息，您需要提示用户提供。如果用户提供了此信息，则需要将此信息存储在数据库中，并告诉用户注册成功。
	存储方法是使用 SQL 语句。您可以使用 SQL 编写插入语句，并且需要生成用户 ID 并将其返回给用户。
	如果用户没有新问题，您应该回复带有 "customer service" 的特殊令牌，以结束任务。
	"""
	query_prompt = """
	您的任务是查询用户信息。您需要从用户那里获得以下信息：
	1. 用户 ID
	2. 用户设置的密码
	如果用户没有提供此信息，则需要提示用户提供。如果用户提供了此信息，那么需要查询数据库。如果用户 ID 和密码匹配，则需要返回用户的信息。
	如果用户没有新问题，您应该回复带有 "customer service" 的特殊令牌，以结束任务。
	"""
	delete_prompt = """
	您的任务是删除用户信息。您需要从用户那里获得以下信息：
	1. 用户 ID
	2. 用户设置的密码
	3. 用户的电子邮件地址
	如果用户没有提供此信息，则需要提示用户提供该信息。
	如果用户提供了这些信息，则需要查询数据库。如果用户 ID 和密码匹配，您需要通知用户验证码已发送到他们的电子邮件，需要进行验证。
	如果用户没有新问题，您应该回复带有 "customer service" 的特殊令牌，以结束任务。
	"""

# 定义 agent

结合上述步骤，定义一个完整的智能客服智能体

	import os
	from dotenv import load_dotenv
	from openai import OpenAI

	class Create_Client:
	def __init__(self):
	# 加载环境变量
	load_dotenv ()
	# 从环境变量中读取 api_key
	self.api_key = os.getenv ('DeepSeek_API_KEY')
	self.base_url = "https://api.siliconflow.cn/v1/"
	self.chat_model = "Pro/deepseek-ai/DeepSeek-V3"

	self.client = OpenAI (
	api_key = self.api_key,
	base_url = self.base_url
	)

	def get_completion (self, prompt):
	response = self.client.chat.completions.create (
	model=self.chat_model, # 模型名称
	messages=[
	{"role": "user", "content": prompt},
	],
	)
	return response.choices [0].message.content


	class Prompts_list:
	def __init__(self):
	self.sys_prompt = """ 你是一个聪明的客服。您将能够根据用户的问题将不同的任务分配给不同的人。您有以下业务线：
	1. 用户注册。如果用户想要执行这样的操作，您应该发送一个带有 "registered workers" 的特殊令牌。并告诉用户您正在调用它。
	2. 用户数据查询。如果用户想要执行这样的操作，您应该发送一个带有 "query workers" 的特殊令牌。并告诉用户您正在调用它。
	3. 删除用户数据。如果用户想执行这种类型的操作，您应该发送一个带有 "delete workers" 的特殊令牌。并告诉用户您正在调用它。
	"""
	self.registered_prompt = """
	您的任务是根据用户信息存储数据。您需要从用户那里获得以下信息：
	1. 用户名、性别、年龄
	2. 用户设置的密码
	3. 用户的电子邮件地址
	如果用户没有提供此信息，您需要提示用户提供。如果用户提供了此信息，则需要将此信息存储在数据库中，并告诉用户注册成功。
	存储方法是使用 SQL 语句。您可以使用 SQL 编写插入语句，并且需要生成用户 ID 并将其返回给用户。
	如果用户没有新问题，您应该回复带有 "customer service" 的特殊令牌，以结束任务。
	"""
	self.query_prompt = """
	您的任务是查询用户信息。您需要从用户那里获得以下信息：
	1. 用户 ID
	2. 用户设置的密码
	如果用户没有提供此信息，则需要提示用户提供。如果用户提供了此信息，那么需要查询数据库。如果用户 ID 和密码匹配，则需要返回用户的信息。
	如果用户没有新问题，您应该回复带有 "customer service" 的特殊令牌，以结束任务。
	"""
	self.delete_prompt = """
	您的任务是删除用户信息。您需要从用户那里获得以下信息：
	1. 用户 ID
	2. 用户设置的密码
	3. 用户的电子邮件地址
	如果用户没有提供此信息，则需要提示用户提供该信息。
	如果用户提供了这些信息，则需要查询数据库。如果用户 ID 和密码匹配，您需要通知用户验证码已发送到他们的电子邮件，需要进行验证。
	如果用户没有新问题，您应该回复带有 "customer service" 的特殊令牌，以结束任务。
	"""

	class SmartAssistant:
	def __init__(self):
	"""
	类定义与初始化
	"""
	customer_Client = Create_Client ()
	customer_prompts = Prompts_list ()
	self.client = customer_Client.client
	self.chat_model = customer_Client.chat_model
	self.system_prompt = customer_prompts.sys_prompt # 通用客服提示
	self.registered_prompt = customer_prompts.registered_prompt # 员工注册专用提示
	self.query_prompt = customer_prompts.query_prompt # 员工查询专用提示
	self.delete_prompt = customer_prompts.delete_prompt # 员工删除专用提示

	# 消息管理系统
	# 使用字典存储 4 个独立的消息历史，每个场景初始化时包含对应的系统提示
	self.messages = {
	"system": [{"role": "system", "content": self.system_prompt}],
	"registered": [{"role": "system", "content": self.registered_prompt}],
	"query": [{"role": "system", "content": self.query_prompt}],
	"delete": [{"role": "system", "content": self.delete_prompt}]
	}

	# 跟踪当前场景（初始为通用客服）
	self.current_assignment = "system"

	def get_response (self, user_input):
	"""
	核心响应方法
	"""
	self.messages [self.current_assignment].append ({"role": "user", "content": user_input})
	while True:
	response = self.client.chat.completions.create (
	model=self.chat_model,
	messages=self.messages [self.current_assignment],
	temperature=0.9,
	stream=False,
	max_tokens=2000,
	)

	ai_response = response.choices [0].message.content
	print (response)
	if "registered workers" in ai_response:
	self.current_assignment = "registered"
	print ("意图识别:",ai_response)
	print ("switch to <registered>")
	self.messages [self.current_assignment].append ({"role": "user", "content": user_input})
	elif "query workers" in ai_response:
	self.current_assignment = "query"
	print ("意图识别:",ai_response)
	print ("switch to <query>")
	self.messages [self.current_assignment].append ({"role": "user", "content": user_input})
	elif "delete workers" in ai_response:
	self.current_assignment = "delete"
	print ("意图识别:",ai_response)
	print ("switch to <delete>")
	self.messages [self.current_assignment].append ({"role": "user", "content": user_input})
	elif "customer service" in ai_response:
	print ("意图识别:",ai_response)
	print ("switch to <customer service>")
	self.messages ["system"] += self.messages [self.current_assignment]
	self.current_assignment = "system"
	return ai_response
	else:
	self.messages [self.current_assignment].append ({"role": "assistant", "content": ai_response})
	return ai_response

	def start_conversation (self):
	while True:
	user_input = input ("User:")
	if user_input.lower () in ['exit', 'quit']:
	print ("Exiting conversation.")
	break
	response = self.get_response (user_input)
	print ("Assistant:", response)

	if __name__ == '__main__':
	assistant = SmartAssistant ()
	assistant.start_conversation ()

get_response 函数解析：

self.current_assignment 标记当前场景

self.messages [self.current_assignment].append ({"role": "user", "content": user_input})

分别存储四个场景下的对话任务

当用户第一次发出请求时，此时 self.current_assignment 默认为 system ，进入主循环后，LLM 判断 user 的意图， response 中携带事先约定好的令牌，但判断某一个令牌 in response 中时，会设置循环执行对应场景的

# 案例一：阅卷智能体

在各种考试中，选择题、判断题、填空题很好阅卷，只需要对比答案和作答是否完全一致即可。但是对于简答题的阅卷，就没这么简单了，通常需要对比意思是否真正答出来了，这通常需要人去阅读考生的作答。现在有了大模型，我们可以让大模型帮助我们给简答题阅卷并生成评价。

	""" 你是一位中国专利代理师考试阅卷专家，
	擅长根据给定的题目和答案为考生生成符合要求的评分和中文评语，
	并按照特定的格式输出。
	你的任务是，根据我输入的考题和答案，针对考生的作答生成评分和中文的评语，并以 JSON 格式返回。
	阅卷标准适当宽松一些，只要考生回答出基本的意思就应当给分。
	答案如果有数字标注，含义是考生如果答出这个知识点，这道题就会得到几分。
	生成的中文评语需要能够被 json.loads () 这个函数正确解析。
	生成的整个中文评语需要用英文的双引号包裹，在被包裹的字符串内部，请用中文的双引号。
	中文评语中不可以出现换行符、转义字符等等。

	输出格式为 JSON:
	{
	"llmgetscore": 0,
	"llmcomments": "中文评语"
	}

	比较学生的回答与正确答案，
	并给出满分为 10 分的评分和中文评语。
	题目：{ques_title}
	答案：{answer}
	学生的回复：{reply}"""

	import json
	import re
	import os
	from dotenv import load_dotenv
	from openai import OpenAI

	class Create_Client:
	def __init__(self):
	# 加载环境变量
	load_dotenv ()
	# 从环境变量中读取 api_key
	self.api_key = os.getenv ('DeepSeek_API_KEY')
	self.base_url = "https://api.siliconflow.cn/v1/"
	self.chat_model = "Pro/deepseek-ai/DeepSeek-V3"

	self.client = OpenAI (
	api_key = self.api_key,
	base_url = self.base_url
	)

	def extract_json_content (text):
	# 这个函数的目标是提取大模型输出内容中的 json 部分，并对 json 中的换行符、首位空白符进行删除
	text = text.replace ("\n","")
	pattern = r"```json (.*?)```"
	matches = re.findall (pattern, text, re.DOTALL)
	if matches:
	return matches [0].strip ()
	return text

	class JsonOutputParser:
	def parse (self, result):
	# 这个函数的目标是把 json 字符串解析成 python 对象
	# 其实这里写的这个函数性能很差，经常解析失败，有很大的优化空间
	try:
	result = extract_json_content (result)
	parsed_result = json.loads (result)
	return parsed_result
	except json.JSONDecodeError as e:
	raise Exception (f"Invalid json output: {result}") from e

	class GradingOpenAI:
	def __init__(self):
	create_client = Create_Client ()
	self.model = create_client.chat_model
	self.client = create_client.client

	self.output_parser = JsonOutputParser ()

	# 注，因 markdown 渲染问题，将提示词单独放在最前
	self.template = prompt.txt

	def create_prompt (self, ques_title, answer, reply):
	return self.template.format (
	ques_title=ques_title,
	answer=answer,
	reply=reply
	)

	def grade_answer (self, ques_title, answer, reply):
	success = False
	while not success:
	# 这里是一个不得已的权宜之计
	# 上面的 json 解析函数不是表现很差吗，那就多生成几遍，直到解析成功
	# 对大模型生成的内容先解析一下，如果解析失败，就再让大模型生成一遍
	try:
	response = self.client.chat.completions.create (
	model=self.model,
	messages=[
	{"role": "system", "content": "你是一位专业的考试阅卷专家。"},
	{"role": "user", "content": self.create_prompt (ques_title, answer, reply)}
	],
	temperature=0.7
	)

	result = self.output_parser.parse (response.choices [0].message.content)
	success = True
	except Exception as e:
	print (f"Error occurred: {e}")
	continue

	return result ['llmgetscore'], result ['llmcomments']

	def run (self, input_data):
	output = []
	for item in input_data:
	score, comment = self.grade_answer (
	item ['ques_title'],
	item ['answer'],
	item ['reply']
	)
	item ['llmgetscore'] = score
	item ['llmcomments'] = comment
	output.append (item)
	return output

	if __name__ == '__main__':
	# 示例输入数据
	input_data = [
	{'ques_title': ' 请解释共有技术特征、区别技术特征、附加技术特征、必要技术特征的含义 ',
	'answer': ' 共有技术特征：与最接近的现有技术共有的技术特征（2.5 分）；区别技术特征：区别于最接近的现有技术的技术特征（2.5 分）；附加技术特征：对所引用的技术特征进一步限定的技术特征，增加的技术特征（2.5 分）；必要技术特征：为解决其技术问题所不可缺少的技术特征（2.5 分）。',
	'fullscore': 10,
	'reply': ' 共有技术特征：与所对比的技术方案相同的技术特征 \n 区别技术特征：与所对比的技术方案相区别的技术特征 \n 附加技术特征：对引用的技术特征进一步限定的技术特征 \n 必要技术特征：解决技术问题必须可少的技术特征 '},
	{'ques_title': ' 请解释前序部分、特征部分、引用部分、限定部分 ',
	'answer': ' 前序部分：独权中，主题 + 与最接近的现有技术共有的技术特征，在其特征在于之前（2.5 分）；特征部分：独权中，与区别于最接近的现有技术的技术特征，在其特征在于之后（2.5 分）；引用部分：从权中引用的权利要求编号及主题（2.5 分）；限定部分：从权中附加技术特征（2.5 分）。',
	'fullscore': 10,
	'reply': ' 前序部分：独立权利要求中与现有技术相同的技术特征 \n 特征部分：独立权利要求中区别于现有技术的技术特征 \n 引用部分：从属权利要求中引用其他权利要求的部分 \n 限定部分：对所引用的权利要求进一步限定的技术特征 '}]
	grading_openai = GradingOpenAI ()
	graded_data = grading_openai.run (input_data)
	print (graded_data)

# Llama-index 实现 Agent 调用工具

LlamaIndex 是一个将大语言模型和外部数据连接在一起的工具。要使用 LlamaIndex 实现让 Agent 调用工具，首先定义工具函数，用来完成 Agent 的任务，然后把工具函数放入 FunctionTool 对象中，供 Agent 能够使用。

注意：大模型会根据函数的注释来判断使用哪个函数来完成任务。所以，注释一定要写清楚函数的功能和返回值。

安装： pip install llama-index
用 LlamaIndex 实现一个简单的 agent demo ，需要导入 ReActAgent 和 Function Tool 。

# ReActAgent

ReActAgent 通过结合推理（Reasoning）和行动（Acting）来创建动态的 LLM Agent 的框架。该方法允许 LLM 模型通过在复杂环境中交替进行推理步骤和行动步骤来更有效地执行任务。ReActAgent 将推理和动作形成了闭环，Agent 可以自己完成给定的任务。

一个典型的 ReActAgent 遵循以下循环：

初始推理：代理首先进行推理步骤，以理解任务、收集相关信息并决定下一步行为。
行动：代理基于其推理采取行动 —— 例如查询 API、检索数据或执行命令。
观察：代理观察行动的结果并收集任何新的信息。
优化推理：利用新信息，代理再次进行推理，更新其理解、计划或假设。
重复：代理重复该循环，在推理和行动之间交替，直到达到满意的结论或完成任务。

# 案例二：Agent 调用工具查看天气

在本次示例中，我们使用 .YML 文件存放所有的 api 接口和 key 信息，封装读取 YML 文件的类 Config_Loder.py ，封装 OurLLM 类 Our_LLM.py ，使用 agent.py 作为 main 文件，项目接口如下

	example/ # 项目根目录
	├── _config.yml # api 配置信息
	├── Config_Loder.py # YML 读取类
	├── Our_LLM.py # OurLLM 类
	└── agent.py # main

# _config.yml

在本次示例中，我们使用 .YML 文件存放所有的 api 接口和 key 信息

	apis:
	LLM:
	# 硅基流动 deepseek api
	api_key: "sk-************"
	base_url: "https://api.siliconflow.cn/v1/"
	chat_model: "Pro/deepseek-ai/DeepSeek-V3"
	weather:
	# 和风天气 api，获取城市代码，获取城市天气
	city_code_url: "https://geoapi.qweather.com/v2/city/lookup"
	city_weather_url: "https://devapi.qweather.com/v7/weather/now"
	weather_key: "************"

# Config_Loder.py

Config_Loder.py 将读取 YML 文件的类封装

	import os
	import yaml
	from pathlib import Path
	from typing import Any, Dict, Optional

	class ConfigLoader:
	"""
	读取 YAML 配置文件

	使用示例：
	>>> config = ConfigLoader.get_instance ()
	>>> api_key = config.get ('qweather.api_key', default='DEMO_KEY')
	"""

	_instance = None # 单例实例
	_config: Dict [str, Any] = {} # 配置缓存
	_loaded = False # 加载状态标记

	def __new__(cls, config_path: str = None):
	if not cls._instance:
	cls._instance = super ().__new__(cls)
	cls._instance._initialize (config_path)
	return cls._instance

	def _initialize (self, config_path: str):
	"""初始化实例，禁止外部直接调用"""
	# 自动检测路径
	if not config_path:
	# 获取当前文件所在目录的父目录
	base_dir = Path (__file__).parent # 根据实际层级调整
	self._config_path = base_dir / "_config.yml"
	else:
	self._config_path = Path (config_path).resolve ()
	self._load_config ()

	def _load_config (self):
	"""执行实际配置加载操作"""
	try:
	# 前置验证
	self._validate_config_file ()

	# 读取文件
	with open (self._config_path, 'r', encoding='utf-8') as f:
	self._config = yaml.safe_load (f) or {}
	self._loaded = True

	except Exception as e:
	self._handle_error (e)
	self._config = {}

	def _validate_config_file (self):
	"""验证配置文件有效性"""
	if not self._config_path.exists ():
	raise FileNotFoundError (f"配置文件 {self._config_path} 不存在")
	if not self._config_path.is_file ():
	raise ValueError (f"{self._config_path} 不是文件")
	if not os.access (self._config_path, os.R_OK):
	raise PermissionError ("缺少读取权限")

	def _handle_error (self, error: Exception):
	"""统一错误处理"""
	error_type = type (error).__name__
	error_map = {
	'FileNotFoundError': f"配置文件不存在：{self._config_path}",
	'PermissionError': f"权限不足，请执行：chmod 600 {self._config_path}",
	'YAMLError': f"YAML 格式错误：{str (error)}",
	'ValueError': f"路径错误：{str (error)}"
	}
	print (error_map.get (error_type, f"未知配置错误：{str (error)}"))

	def get (self, key_path: str, default: Any = None) -> Optional [Any]:
	"""
	安全获取配置项（支持点分嵌套路径）

	参数：
	key_path: 配置项路径，例如 'database.host'
	default: 当配置项不存在时的默认值

	返回：
	配置值或默认值
	"""
	if not self._loaded:
	return default

	keys = key_path.split ('.')
	value = self._config

	try:
	for key in keys:
	if isinstance (value, dict):
	value = value [key]
	else:
	return default
	return value
	except (KeyError, TypeError):
	return default

	@classmethod
	def get_instance (cls) -> 'ConfigLoader':
	"""获取单例实例"""
	if not cls._instance:
	cls._instance = cls () # 使用默认路径
	return cls._instance

	def reload (self) -> bool:
	"""重新加载配置文件"""
	try:
	self._load_config ()
	return True
	except Exception:
	return False

	@property
	def is_loaded (self) -> bool:
	"""检查配置是否成功加载"""
	return self._loaded

	def __contains__(self, key_path: str) -> bool:
	"""检查配置项是否存在"""
	return self.get (key_path, self) is not self # 使用哨兵值检测

	# 使用示例
	if __name__ == "__main__":
	# 初始化配置（自动加载）
	config = ConfigLoader.get_instance ()

	# 检查加载状态
	if not config.is_loaded:
	print ("配置加载失败，请检查错误信息")
	exit (1)

	# 获取配置项
	api_key = config.get ('apis.LLM.api_key')
	print (api_key)
	# db_host = config.get ('database.host', 'localhost')

	# 检查配置项存在性
	if 'logging.level' in config:
	print ("日志配置已存在")

	# 重新加载配置
	if config.reload ():
	print ("配置重载成功")

# Our_LLM.py

定义 OurLLM 类，继承自 CustomLLM 基类

	from openai import OpenAI
	from pydantic import Field # 导入 Field，用于 Pydantic 模型中定义字段的元数据
	from llama_index.core.llms import (
	CustomLLM,
	CompletionResponse,
	LLMMetadata,
	)
	from llama_index.core.llms.callbacks import llm_completion_callback
	from typing import List, Any, Generator

	from Config_Loader import ConfigLoader

	# 定义 OurLLM 类，继承自 CustomLLM 基类
	class OurLLM (CustomLLM):
	api_key: str = Field (default=ConfigLoader.get_instance ().get ('apis.LLM.api_key'))
	base_url: str = Field (default=ConfigLoader.get_instance ().get ('apis.LLM.base_url'))
	model_name: str = Field (default=ConfigLoader.get_instance ().get ('apis.LLM.chat_model'))
	client: OpenAI = Field (default=None, exclude=True)

	def __init__(self, **kwargs):
	# 先调用父类初始化
	super ().__init__(**kwargs)

	# 自动初始化 client（如果字段未设置）
	if not self.client:
	self.client = OpenAI (
	api_key=self.api_key,
	base_url=self.base_url
	)

	@property
	def metadata (self) -> LLMMetadata:
	"""Get LLM metadata."""
	return LLMMetadata (
	model_name=self.model_name,
	)

	@llm_completion_callback ()
	def complete (self, prompt: str, **kwargs: Any) -> CompletionResponse:
	response = self.client.chat.completions.create (model=self.model_name, messages=[{"role": "user", "content": prompt}])
	if hasattr (response, 'choices') and len (response.choices) > 0:
	response_text = response.choices [0].message.content
	return CompletionResponse (text=response_text)
	else:
	raise Exception (f"Unexpected response format: {response}")

	@llm_completion_callback ()
	def stream_complete (
	self, prompt: str, **kwargs: Any
	) -> Generator [CompletionResponse, None, None]:
	response = self.client.chat.completions.create (
	model=self.model_name,
	messages=[{"role": "user", "content": prompt}],
	stream=True
	)

	try:
	for chunk in response:
	chunk_message = chunk.choices [0].delta
	if not chunk_message.content:
	continue
	content = chunk_message.content
	yield CompletionResponse (text=content, delta=content)

	except Exception as e:
	raise Exception (f"Unexpected response format: {e}")

# agent.py

agent.py 作为入口函数，实现调用 LLM 和工具函数

	import requests
	from Our_LLM import OurLLM
	from Config_Loader import ConfigLoader

	from llama_index.core.agent import ReActAgent
	from llama_index.core.tools import FunctionTool

	def multiply (a: float, b: float) -> float:
	"""Multiply two numbers and returns the product"""
	return a * b


	def add (a: float, b: float) -> float:
	"""Add two numbers and returns the sum"""
	return a + b

	def get_weather (city: str) -> int:
	"""
	Gets the weather temperature of a specified city.

	Args:
	city (str): The name or abbreviation of the city.

	Returns:
	int: The temperature of the city. Returns 20 for 'wuhan' (武汉),
	30 for 'beijing' (北京), and -1 for unknown cities.
	"""
	# 调用 GeoAPI 获取城市代码，再使用和风 API 获取天气
	# 定义 API 请求的 URL 和请求头
	config = ConfigLoader.get_instance ()
	city_code_url = config.get ('apis.weather.city_code_url')
	city_weather_url = config.get ('apis.weather.city_weather_url')
	weather_key = config.get ('apis.weather.weather_key')

	city_code_url = f'{city_code_url}?location={city}&number=1'
	headers = {
	'X-QW-Api-Key': weather_key,
	'Accept-Encoding': 'gzip, deflate, br'
	}

	# 发送 GET 请求
	city_response = requests.get (city_code_url, headers=headers)

	# 检查 HTTP 状态码
	if city_response.status_code != 200:
	print (f"请求失败，HTTP 状态码：{city_response.status_code}")
	return -1
	try:
	response_data = city_response.json ()

	# 提取并返回城市代码
	city_code = response_data ['location'][0]['id']
	print (f"匹配到的城市代码：{city_code}")
	except requests.exceptions.JSONDecodeError:
	print ("错误：响应内容无法解析为 JSON 格式")
	return -1

	city_weather_url = f'{city_weather_url}?location={city_code}'
	# 发送 GET 请求
	weather_response = requests.get (city_weather_url, headers=headers)

	# 检查 HTTP 状态码
	if weather_response.status_code != 200:
	print (f"请求失败，HTTP 状态码：{weather_response.status_code}")
	return -1
	try:
	response_data = weather_response.json ()

	# 提取并返回气温
	city_weather = response_data ['now']['temp']
	print (f"匹配到的气温：{city_weather}")
	return city_weather
	except requests.exceptions.JSONDecodeError:
	print ("错误：响应内容无法解析为 JSON 格式")
	return -1


	def main ():

	multiply_tool = FunctionTool.from_defaults (fn=multiply)
	add_tool = FunctionTool.from_defaults (fn=add)
	weather_tool = FunctionTool.from_defaults (fn=get_weather)

	llm = OurLLM ()
	# 创建 ReActAgent 实例
	agent = ReActAgent.from_tools ([multiply_tool, add_tool, weather_tool], llm=llm, verbose=True)
	response = agent.chat ("北京今天天气怎么样？")
	print (response)

	if __name__ == "__main__":
	main ()