在我的上一篇博客：基于谷歌ADK的 智能产品推荐系统(1): 功能简介-CSDN博客 中我们介绍了个性化购物 Agent 项目，该项目展示了一个强大的框架，旨在模拟和实现在线购物环境中的智能导购。它不仅仅是一个简单的聊天机器人，更是一个集成了环境模拟、自然语言理解、搜索引擎、以及复杂交互逻辑的综合系统。该项目不仅是电子商务领域 AI 应用的一个优秀范例，也为研究和开发更复杂的交互式 Agent 提供了宝贵的参考和实践基础。大家可以在github上访问该项目: adk-samples/python/agents/personalized-shopping at main · google/adk-samples · GitHub
核心架构概览
该项目的架构可以概括为：
- 表示层/环境层 (WebShop Simulation): 一个模拟的 Web 购物环境，用户（或 Agent）可以在其中执行搜索、浏览商品、查看详情等操作。
- 感知与行动层 (Tools): Agent 通过 search 工具从环境中获取信息，通过 click 工具在环境中执行操作。
- 决策层 (LLM Agent): 基于 LLM 的 Agent，它接收用户指令和环境观察，决策下一步的行动。
- 数据层 (Product Data & Search Index): 包含大量产品信息，并通过搜索引擎建立索引以支持快速检索。
参考项目 README.md 中的架构图，我们可以更直观地理解各组件的协同工作。
该项目的目录结构：
web_agent_text_env.py 功能详解
shared_libraries/web_agent_site/envs/web_agent_text_env.py 是个性化购物 Agent 项目的核心组件之一，它负责构建和管理一个模拟的 Web 购物环境，它是 OpenAI Gym 环境 (gym.Env) 的一个实现，这个环境模拟了一个简化的电子商务网站（WebShop）。Agent 在这个环境中进行交互，执行搜索、点击等操作，并接收环境的反馈来完成购物任务。可以将它理解为 Agent 进行“训练”和“实践”的操场。
# web_agent_text_env.py
from collections import defaultdict
import json
import random
import string
import time
from bs4 import BeautifulSoup
from bs4.element import Comment
from flask import Flask # SimServer 内部模拟 Flask 应用上下文，并非实际运行 Flask 服务器
import gym
from gym.envs.registration import register
import numpy as np
import torch
from ..engine.engine import (
ACTION_TO_TEMPLATE,
BACK_TO_SEARCH,
END_BUTTON,
NEXT_PAGE,
PREV_PAGE,
get_product_per_page,
get_top_n_product_from_keywords,
init_search_engine,
load_products,
map_action_to_html,
parse_action,
)
from ..engine.goal import get_goals, get_reward
from ..utils import (
DEFAULT_FILE_PATH,
FEAT_CONV,
FEAT_IDS,
random_idx,
)
app = Flask(__name__) # 用于 SimServer 模拟 Flask 应用上下文
# WebAgentTextEnv 类：Gym 环境的实现
# 继承自 gym.Env，遵循 OpenAI Gym 的环境接口规范。
# 负责构建和管理一个模拟的 Web 购物环境，Agent 在此环境中交互。
class WebAgentTextEnv(gym.Env):
"""Gym environment for Text mode of WebShop environment"""
# 初始化 WebAgentTextEnv 环境
def __init__(
self,
observation_mode="html", # Agent 从环境中获取的观测信息的格式 ('html', 'text', 'text_rich', 'url')
file_path=DEFAULT_FILE_PATH, # 指向包含产品数据的 JSON 文件
server=None, # 可以传入一个已有的 SimServer 实例，或内部创建
**kwargs, # 其他可选参数，如图像加载、历史记录长度等
):
"""Constructor for text environment
Arguments:
observation_mode (`str`) -- ['html' | 'text'] (default 'html')
file_path (`str`) -- path to the items_shuffle.json file (default DEFAULT_FILE_PATH)
server (`SimServer`) -- SimServer instance (default None)
**kwargs:
get_image (`bool`): whether to load image features (default 0)
filter_goals (`func`): function to filter goals (default None)
limit_goals (`int`): limit number of goals (default -1)
num_products (`int`): number of products to load (default None)
human_goals (`bool`): whether to load human goals (default 0)
session (`str`): session id (default None, randomly generated)
session_prefix (`str`): session id prefix (default None)
show_attrs (`bool`): whether to show attributes on item page (default False)
num_prev_obs (`int`): number of previous observations to include in state (default 0)
num_prev_actions (`int`): number of previous actions to include in state (default 0)
"""
super(WebAgentTextEnv, self).__init__()
self.observation_mode = observation_mode
self.kwargs = kwargs
self.file_path = file_path
self.base_url = "http://127.0.0.1:3000" # 模拟的基础 URL
# SimServer 实例，负责模拟后端 Web 服务器的逻辑
self.server = (
SimServer(
self.base_url,
self.file_path,
self.kwargs.get("filter_goals"),
self.kwargs.get("limit_goals", -1),
self.kwargs.get("num_products"),
self.kwargs.get("human_goals"),
self.kwargs.get("show_attrs", False),
)
if server is None
else server
)
# SimBrowser 实例，负责模拟浏览器的行为
self.browser = SimBrowser(self.server)
# 会话管理：用于跟踪和区分不同的用户交互会话
self.session = self.kwargs.get("session")
self.session_prefix = self.kwargs.get("session_prefix")
# 可选：如果启用图像功能，加载预计算的图像特征
if self.kwargs.get("get_image", 0):
self.feats = torch.load(FEAT_CONV)
self.ids = torch.load(FEAT_IDS)
self.ids = {url: idx for idx, url in enumerate(self.ids)}
# 交互历史：存储先前若干步的观测和动作
self.prev_obs = []
self.prev_actions = []
self.num_prev_obs = self.kwargs.get("num_prev_obs", 0)
self.num_prev_actions = self.kwargs.get("num_prev_actions", 0)
self.reset() # 初始化结束时调用 reset，准备第一个会话
# step 方法：执行 Agent 的一个动作并返回结果
def step(self, action):
"""Takes an action, updates WebShop environment, and returns (observation, reward, done, info)
Arguments:
action (`str`): An action should be of the following structure:
- search[keywords]
- click[value]
If action not valid, perform nothing.
"""
info = None # 附加信息，当前版本为 None
self.get_available_actions() # 获取当前页面可执行的动作
# 动作解析：将输入的动作字符串解析为动作名称和参数
action_name, action_arg = parse_action(action)
if action_arg is not None:
action_arg = action_arg.lower()
# 动作执行
if action_name == "search" and action_arg is not None and action_arg != "":
# 如果是搜索动作，调用 SimBrowser 的 search 方法
status = self.browser.search(action_arg)
elif (
action_name == "click"
and action_arg in self.text_to_clickable.keys() # 验证点击目标是否存在
and action_arg != "search"
):
# 如果是点击动作，调用 SimBrowser 的 click 方法
status = self.browser.click(action_arg, self.text_to_clickable)
else:
# 无效动作，不执行任何操作
status = dict(reward=0, done=False)
# 更新观测和状态
ob = self.observation # 获取新的观测
text_list = [ob]
# 将当前动作和观测添加到交互历史中
self.prev_actions.append(action)
# 根据配置，将历史观测和动作与当前观测拼接成新的状态
for i in range(1, 1 + max(self.num_prev_obs, self.num_prev_actions)):
if len(self.prev_actions) >= i and self.num_prev_actions >= i:
text_list.append(self.prev_actions[-i])
if len(self.prev_obs) >= i and self.num_prev_obs >= i:
text_list.append(self.prev_obs[-i])
state = " [SEP] ".join(text_list[::-1]) # 使用 [SEP] 分隔历史信息
self.prev_obs.append(ob)
# 返回新的状态、奖励、是否结束标志和附加信息
return state, status["reward"], status["done"], info
# get_available_actions 方法：获取当前页面可用的动作列表
def get_available_actions(self):
"""Returns list of available actions at the current step"""
# 使用 BeautifulSoup 解析当前页面的 HTML
html_obj = self._parse_html()
# 查找可点击元素：搜索框、按钮、产品链接、购买选项
search_bar = html_obj.find(id="search_input")
has_search_bar = True if search_bar is not None else False
buttons = html_obj.find_all(class_="btn")
product_links = html_obj.find_all(class_="product-link")
buying_options = html_obj.select('input[type="radio"]')
# 构建可点击元素的文本到元素对象的映射，用于验证点击动作
self.text_to_clickable = {
f"{b.get_text()}".lower(): b for b in buttons + product_links
}
for opt in buying_options:
opt_value = opt.get("value")
self.text_to_clickable[f"{opt_value}"] = opt
# 返回包含搜索框信息和可点击元素文本列表的字典
return dict(
has_search_bar=has_search_bar,
clickables=list(self.text_to_clickable.keys()),
)
# get_image 方法：提取当前产品页面的图像特征 (如果启用图像功能)
def get_image(self):
"""Scrape image from page HTML and return as a list of pixel values"""
html_obj = self._parse_html(self.browser.page_source)
image_url_element = html_obj.find(id="product-image")
if image_url_element is not None:
image_url = image_url_element["src"]
if image_url in self.ids: # 在预加载的图像特征库中查找
image_idx = self.ids[image_url]
image_features = self.feats[image_idx]
return image_features
return torch.zeros(512) # 未找到则返回零向量
# get_instruction_text 方法：从当前页面 HTML 中提取指令文本 (购物目标描述)
def get_instruction_text(self):
"""Get corresponding instruction text for current environment session"""
html_obj = self._parse_html(self.browser.page_source)
instruction_text_element = html_obj.find(id="instruction-text")
if instruction_text_element and instruction_text_element.h4:
return instruction_text_element.h4.text
return "" # 如果未找到指令文本元素，返回空字符串
# _parse_html 方法：辅助方法，将 HTML 字符串包装成 BeautifulSoup 对象
def _parse_html(self, html=None):
"""Returns web request result wrapped in BeautifulSoup object
Arguments:
url (`str`): If no url or html is provided, use the current
observation (HTML) for parsing.
"""
if html is None:
html = self.state["html"] # 默认使用当前状态的 HTML
html_obj = BeautifulSoup(html, "html.parser")
return html_obj
# observation 属性：根据配置的 observation_mode 返回不同格式的观测信息
@property
def observation(self):
"""Compiles state into either the `html` or `text` observation mode"""
html = self.state["html"]
if self.observation_mode == "html":
return html
elif self.observation_mode == "text":
return self.convert_html_to_text(html, simple=True)
elif self.observation_mode == "text_rich":
return self.convert_html_to_text(html, simple=False)
elif self.observation_mode == "url":
return self.state["url"]
else:
raise ValueError(f"Observation mode {self.observation_mode} not supported.")
# state 属性：返回包含环境完整信息的字典 (URL, HTML, 指令文本)
@property
def state(self):
"""State that includes all information.
The actual observation are likely to be a subset or reduced form of the
state.
"""
return dict(
url=self.browser.current_url,
html=self.browser.page_source,
instruction_text=self.instruction_text,
)
# convert_html_to_text 方法：将 HTML 源码转换为文本表示
def convert_html_to_text(self, html, simple=False):
"""Strip HTML of tags and add separators to convert observation into simple mode"""
texts = self._parse_html(html).findAll(text=True)
# 使用 tag_visible 过滤掉不可见元素（如<script>, <style>等）的文本
visible_texts = filter(tag_visible, texts)
if simple:
# 'text' 模式：将所有可见文本去除首尾空格后，用 " [SEP] " 连接
return " [SEP] ".join(t.strip() for t in visible_texts if t.strip())
else:
# 'text_rich' 模式：为不同类型的 HTML 元素添加特定的标记
observation = ""
for t in visible_texts:
t_stripped = t.strip()
if not t_stripped: # 跳过空行或纯空格
continue
parent_tag = t.parent
parent_name = parent_tag.name
processed_t = t_stripped # 默认为原始文本
if parent_name == "button": # 按钮
processed_t = f"[button] {t_stripped} [button_]"
elif parent_name == "label": # 购买选项 (如颜色、尺寸的标签)
# 检查选项是否已被选中 (通过URL判断)
# 注意: 这里的 self.state["url"] 可能不是最新，因为convert_html_to_text可能被用于任意html
# 更好的做法是检查父 input 元素的 checked 属性，但这需要更复杂的DOM遍历
# 当前实现可能不完全准确反映“已点击”状态，除非URL中包含选项信息
if f'"{t_stripped}"' in self.browser.current_url: # 尝试使用当前浏览器URL
processed_t = f" [clicked button] {t_stripped} [clicked button_]"
# 优先显示已点击的选项信息
# observation = f"You have clicked {t_stripped}.\n" + observation
else:
processed_t = f" [button] {t_stripped} [button_]"
elif parent_tag.get("class") and "product-link" in parent_tag.get("class"): # 产品链接 (ASIN)
# 检查产品链接是否曾被点击过 (通过会话中记录的ASINs判断)
# 需要访问 self.server.user_sessions[self.session]["asins"]，但这会增加耦合
# 简化：暂时不在此处标记已点击的产品链接，因为链接本身通常不会改变外观表示已点击
processed_t = f"\n[button] {t_stripped} [button_]"
# else: # 常规、不可点击文本，保持原样
# processed_t = t_stripped
observation += processed_t + "\n"
return observation.strip() # 移除末尾的换行符
# reset 方法：重置环境到一个新的初始状态
def reset(self, session=None, instruction_text=None):
"""Create a new session and reset environment variables"""
session_int = None
# 会话 ID 管理
if session is not None:
self.session = str(session)
if isinstance(session, int):
session_int = session
else:
self.session = "".join(random.choices(string.ascii_lowercase, k=10)) # 随机生成会话 ID
if self.session_prefix is not None:
self.session = self.session_prefix + self.session # 添加会话 ID 前缀
init_url = f"{self.base_url}/{self.session}" # 构建初始页面的 URL (通常是搜索首页)
# 调用 SimBrowser 的 get 方法加载初始页面，并传递会话信息
self.browser.get(init_url, session_id=self.session, session_int=session_int)
self.text_to_clickable = None # 清空上一会话的可点击元素映射
# 获取指令文本 (购物目标)
self.instruction_text = (
self.get_instruction_text() # 从新加载的页面获取
if instruction_text is None # 如果外部未提供
else instruction_text
)
obs = self.observation # 获取初始观测
# 清空并初始化交互历史
self.prev_obs = [obs]
self.prev_actions = []
return obs, None # 返回初始观测和附加信息
# render 方法：Gym 标准方法，当前实现为空操作
def render(self, mode="human"):
pass
# close 方法：Gym 标准方法，当前实现为空操作
def close(self):
pass
# tag_visible 函数：辅助函数，用于 BeautifulSoup 文本提取时判断 HTML 元素是否可见
# 忽略 <style>, <script>, <head>, <title>, <meta> 标签及 HTML 注释
def tag_visible(element):
ignore = {"style", "script", "head", "title", "meta", "[document]"}
return element.parent.name not in ignore and not isinstance(element, Comment)
# SimServer 类：轻量级 Web 服务器模拟器
# 负责模拟 WebShop 应用的后端逻辑，处理来自 SimBrowser 的“请求”。
class SimServer:
"""Lightweight simulator of WebShop Flask application for generating HTML observations"""
# 初始化 SimServer
def __init__(
self,
base_url,
file_path, # 产品数据文件路径
filter_goals=None, # 可选的目标筛选函数
limit_goals=-1, # 限制可用目标数量
num_products=None, # 控制加载和索引的产品数量
human_goals=0, # 是否加载人类标注的目标
show_attrs=False, # 是否在商品页面显示额外属性
):
"""Constructor for simulated server serving WebShop application
Arguments:
filter_goals (`func`) -- Select specific goal(s) for consideration based on
criteria of custom function
limit_goals (`int`) -- Limit to number of goals available
num_products (`int`) -- Number of products to search across
human_goals (`bool`) -- If true, load human goals; otherwise, load synthetic
goals
"""
# 加载所有产品信息、产品字典、产品价格
self.base_url = base_url
self.all_products, self.product_item_dict, self.product_prices, _ = (
load_products(
filepath=file_path,
num_products=num_products,
human_goals=human_goals,
)
)
# 初始化 Pyserini 搜索引擎
self.search_engine = init_search_engine(num_products=num_products)
# 加载或生成购物目标
self.goals = get_goals(self.all_products, self.product_prices, human_goals)
self.show_attrs = show_attrs # 是否在商品页显示属性的配置
# 固定随机种子以便复现目标打乱顺序
random.seed(233)
random.shuffle(self.goals)
# 如果提供了目标筛选函数，则应用筛选
if filter_goals is not None:
self.goals = [
goal for (i, goal) in enumerate(self.goals) if filter_goals(i, goal)
]
# 如果设置了目标数量限制，则随机选择指定数量的目标
if limit_goals != -1 and limit_goals < len(self.goals):
self.weights = [goal["weight"] for goal in self.goals] # 目标权重，用于抽样
self.cum_weights = [0] + np.cumsum(self.weights).tolist()
idxs = []
while len(idxs) < limit_goals:
idx = random_idx(self.cum_weights)
if idx not in idxs:
idxs.append(idx)
self.goals = [self.goals[i] for i in idxs]
print(f"Loaded {len(self.goals)} goals.")
# 会话存储：字典，用于存储每个会话的状态信息
self.weights = [goal["weight"] for goal in self.goals]
self.cum_weights = [0] + np.cumsum(self.weights).tolist()
self.user_sessions = dict()
# 性能计时变量
self.search_time = 0
self.render_time = 0
self.sample_time = 0
# 特殊变量，允许外部临时覆盖当前会话的指令文本显示 (例如在搜索时)
self.assigned_instruction_text = None
# 模拟 Flask 路由：首页/搜索页
# @app.route("/", methods=["GET", "POST"]) # 装饰器仅为示意，实际通过 receive 调用
def index(self, session_id, **kwargs):
"""Redirect to the search page with the given session ID"""
# 渲染搜索页面的 HTML
html = map_action_to_html(
"start", # 动作类型
session_id=session_id,
instruction_text=kwargs["instruction_text"], # 页面显示的指令文本
)
url = f"{self.base_url}/{session_id}" # 页面 URL
return html, url
# 模拟 Flask 路由：搜索结果页
# @app.route("/", methods=["GET", "POST"])
def search_results(self, session_id, **kwargs):
"""Initialize session and return the search results page"""
session = self.user_sessions[session_id] # 获取当前会话信息
keywords = kwargs["keywords"] # 搜索关键词
assert isinstance(keywords, list)
page = 1 if "page" not in kwargs else kwargs["page"] # 当前页码
# 更新会话状态
session["page"] = page
session["keywords"] = keywords
session["actions"]["search"] += 1 # 记录搜索动作次数
session["asin"] = None # 清空之前选择的商品
session["options"] = {} # 清空之前选择的选项
# 调用搜索引擎执行搜索
old_time = time.time()
top_n_products = get_top_n_product_from_keywords(
keywords,
self.search_engine,
self.all_products,
self.product_item_dict,
)
self.search_time += time.time() - old_time # 记录搜索耗时
# 获取当前页的产品列表
products_on_page = get_product_per_page(top_n_products, page)
keywords_url_string = "+".join(keywords)
url = (
f"{self.base_url}/search_results/{session_id}/"
f"{keywords_url_string}/{page}"
)
# 渲染搜索结果页面的 HTML
old_time = time.time()
html = map_action_to_html(
"search",
session_id=session_id,
products=products_on_page,
keywords=session["keywords"],
page=page,
total=len(top_n_products), # 总结果数
instruction_text=self.assigned_instruction_text or session['goal']['instruction_text'], # 优先使用 assigned_instruction_text
)
self.render_time += time.time() - old_time # 记录渲染耗时
return html, url
# 模拟 Flask 路由：商品详情页
# @app.route("/", methods=["GET", "POST"])
def item_page(self, session_id, **kwargs):
"""Render and return the HTML for a product item page"""
session = self.user_sessions[session_id]
clickable_name = kwargs["clickable_name"] # 被点击的元素名称 (通常是商品 ASIN 或购买选项)
text_to_clickable = kwargs["text_to_clickable"] # 当前页面的可点击元素映射
clickable_element = text_to_clickable[clickable_name]
# 更新会话状态，记录选择的商品或选项
if (
clickable_element.get("class") is not None
and clickable_element.get("class")[0] == "product-link" # 如果点击的是产品链接
):
session["asin"] = clickable_name.upper() # 记录商品 ASIN
session["actions"]["asin"] += 1
session["asins"].add(session["asin"]) # 将 ASIN 添加到已查看列表
elif clickable_element.get("name") is not None: # 如果点击的是购买选项 (radio button)
option_key = clickable_element["name"].lower() # 选项类型 (如 'color', 'size')
session["options"][option_key] = clickable_name # 记录选择的选项值
session["actions"]["options"] += 1
# 获取商品信息并渲染商品详情页 HTML
product_info = self.product_item_dict[session["asin"]]
keywords_url_string = "+".join(session["keywords"])
option_string = json.dumps(session["options"]) # 将选项字典转为 JSON 字符串用于 URL
url = (
f"{self.base_url}/item_page/{session_id}/"
f'{session["asin"]}/{keywords_url_string}/'
f'{session["page"]}/{option_string}'
)
html = map_action_to_html(
"click", # 动作类型
session_id=session_id,
product_info=product_info,
keywords=session["keywords"],
page=session["page"],
asin=session["asin"],
options=session["options"],
instruction_text=self.assigned_instruction_text or session['goal']['instruction_text'],
show_attrs=self.show_attrs, # 是否显示额外属性
)
return html, url
# 模拟 Flask 路由：商品子信息页 (描述、特性、评论)
# @app.route("/", methods=["GET", "POST"])
def item_sub_page(self, session_id, **kwargs):
"""Render and return the HTML for a product's sub page (i.e. description, features)"""
session = self.user_sessions[session_id]
clickable_name = kwargs["clickable_name"] # 被点击的子页面类型 (如 'Description')
# 标准化 clickable_name
for k_template in ACTION_TO_TEMPLATE:
if clickable_name.lower() == k_template.lower():
clickable_name = k_template
break
# 获取商品信息并渲染商品子信息页 HTML
product_info = self.product_item_dict[session["asin"]]
session["actions"][clickable_name] += 1 # 记录查看子页面的动作次数
keywords_url_string = "+".join(session["keywords"])
url = (
f"{self.base_url}/item_sub_page/{session_id}/"
f'{session["asin"]}/{keywords_url_string}/{session["page"]}/'
f'{clickable_name}/{json.dumps(session["options"])}' # options也传入URL
)
html = map_action_to_html(
f"click[{clickable_name}]", # 动作类型，如 click[Description]
session_id=session_id,
product_info=product_info,
keywords=session["keywords"],
page=session["page"],
asin=session["asin"],
options=session["options"],
instruction_text=self.assigned_instruction_text or session['goal']['instruction_text'],
)
return html, url
# 模拟 Flask 路由：购买完成页
# @app.route("/", methods=["GET", "POST"])
def done(self, session_id, **kwargs):
"""Render and return HTML for done page"""
session = self.user_sessions[session_id]
goal = session["goal"] # 当前会话的购物目标
# 获取购买的商品信息
purchased_product = self.product_item_dict.get(session["asin"])
if not purchased_product: # 如果ASIN无效或未选择商品
return "Error: Product not found or not selected.", "", 0.0
session["actions"]["purchase"] += 1 # 记录购买动作次数
price = self.product_prices.get(session["asin"])
# 计算奖励
reward, info = get_reward(
purchased_product,
goal,
price=price,
options=session["options"],
verbose=True, # 获取详细的奖励信息
)
# 更新会话状态，标记为已完成并记录奖励
session["verbose_info"] = info
session["done"] = True
session["reward"] = reward
url = (
f"{self.base_url}/done/{session_id}/"
f'{session["asin"]}/{json.dumps(session["options"])}'
)
# 渲染购买完成页 HTML
html = map_action_to_html(
f"click[{END_BUTTON}]", # 动作类型 click[Buy Now]
session_id=session_id,
reward=reward,
asin=session["asin"],
options=session["options"],
reward_info=info, # 传递详细奖励信息给模板
goal=goal, # 传递目标信息给模板
instruction_text=self.assigned_instruction_text or session['goal']['instruction_text'],
)
return html, url, reward # 返回 HTML, URL 和奖励值
# receive 方法：SimServer 的核心入口，模拟接收 HTTP 请求并返回响应
def receive(self, session_id, current_url, session_int=None, **kwargs):
"""Map action to the corresponding page"""
status = dict(reward=0.0, done=False) # 初始化返回状态
# 使用 Flask 应用上下文，主要为了 render_template_string 等函数能正确工作
with app.app_context(), app.test_request_context():
# 会话初始化与目标分配
if session_id not in self.user_sessions: # 如果是新会话
idx = (
session_int
if (session_int is not None and isinstance(session_int, int))
else random_idx(self.cum_weights) # 随机选择一个目标
)
goal = self.goals[idx]
instruction_text = goal["instruction_text"]
# 初始化会话存储
self.user_sessions[session_id] = {
"goal": goal,
"done": False,
"keywords": None,
"page": None,
"asin": None,
"asins": set(),
"options": dict(),
"actions": defaultdict(int),
}
else: # 已有会话
instruction_text = self.user_sessions[session_id]["goal"]["instruction_text"]
# 如果设置了 assigned_instruction_text (通常由 search 工具设置)，则优先使用它
# 这是一个有些 hacky 的方式来动态更新页面显示的指令
if self.assigned_instruction_text is not None:
current_instruction_text = self.assigned_instruction_text
else:
current_instruction_text = self.user_sessions[session_id]["goal"]["instruction_text"]
# 更新会话中的目标指令文本 (如果被 assigned_instruction_text 覆盖)
# self.user_sessions[session_id]["goal"]["instruction_text"] = current_instruction_text
session = self.user_sessions[session_id]
# 请求分发逻辑
if not kwargs: # 如果没有额外参数 (通常是加载初始页面)
kwargs["instruction_text"] = current_instruction_text
html, url = self.index(session_id, **kwargs)
elif "keywords" in kwargs: # 如果是搜索请求
kwargs["instruction_text"] = current_instruction_text # 传递指令文本给 search_results
html, url = self.search_results(session_id, **kwargs)
elif "clickable_name" in kwargs: # 如果是点击请求
clickable_name_lower = kwargs["clickable_name"].lower()
kwargs["instruction_text"] = current_instruction_text # 传递指令文本
if clickable_name_lower == END_BUTTON.lower(): # 点击 "Buy Now"
html, url, reward = self.done(session_id, **kwargs)
status["reward"] = reward
status["done"] = True
elif clickable_name_lower == BACK_TO_SEARCH.lower(): # 点击 "Back to Search"
# 重置到搜索首页，清空 kwargs 以调用 index
html, url = self.index(session_id, instruction_text=current_instruction_text)
# 重置会话中的搜索相关状态，但不重置目标
session.update({
"keywords": None, "page": None, "asin": None,
"asins": set(), "options": dict(),
# "actions": defaultdict(int) # 保留动作计数
})
elif (
clickable_name_lower == NEXT_PAGE.lower()
and self.get_page_name(current_url) == "search_results" # 在搜索结果页点击 "Next >"
):
html, url = self.search_results(
session_id,
keywords=session["keywords"],
page=session["page"] + 1,
**kwargs, # 传递 instruction_text
)
elif (
clickable_name_lower == PREV_PAGE.lower()
and self.get_page_name(current_url) == "search_results" # 在搜索结果页点击 "< Prev"
):
html, url = self.search_results(
session_id,
keywords=session["keywords"],
page=session["page"] - 1,
**kwargs, # 传递 instruction_text
)
elif (
clickable_name_lower == PREV_PAGE.lower() # 点击 "< Prev"
and self.get_page_name(current_url) == "item_sub_page" # 在商品子页面
):
# 返回商品主详情页
html, url = self.item_page(session_id, **kwargs)
elif (
clickable_name_lower == PREV_PAGE.lower() # 点击 "< Prev"
and self.get_page_name(current_url) == "item_page" # 在商品主详情页
):
# 返回之前的搜索结果页
html, url = self.search_results(
session_id,
keywords=session["keywords"],
page=session["page"],
**kwargs, # 传递 instruction_text
)
elif clickable_name_lower in [k.lower() for k in ACTION_TO_TEMPLATE]: # 点击 "Description", "Features", "Reviews"
html, url = self.item_sub_page(session_id, **kwargs)
else: # 其他点击 (通常是商品链接或购买选项)
html, url = self.item_page(session_id, **kwargs)
else: # 未知情况，理论上不应发生
html, url = "Error: Unknown request", current_url
return html, url, status
# get_page_name 方法：辅助方法，根据 URL 字符串判断当前属于哪个类型的页面
def get_page_name(self, url):
"""Determine which page (i.e. item_page, search_results) the given URL is pointing at."""
if url is None:
return None
page_names = ["search_results", "item_page", "item_sub_page", "done"]
for page_name in page_names:
if page_name in url:
return page_name
return "" # index page (首页)
# SimBrowser 类：模拟浏览器行为
# 封装了与 SimServer 交互的逻辑，模拟用户在浏览器中的操作。
class SimBrowser:
"""Simulated browser for rendering the HTML source of WebShop environment pages."""
# 初始化 SimBrowser
def __init__(self, server):
self.server = server # 持有 SimServer 实例
self.current_url = None # 当前 URL
self.page_source = None # 当前页面的 HTML 源码
self.session_id = None # 当前会话 ID
# get 方法：模拟浏览器加载一个新 URL
def get(self, url, session_id=None, session_int=None):
"""Set browser variables to corresponding link, page HTML for URL"""
self.session_id = url.split("/")[-1] if session_id is None else session_id
# 调用 SimServer 的 receive 方法获取页面内容 (不带额外参数，用于初始加载)
self.page_source, self.current_url, _ = self.server.receive(
self.session_id, self.current_url, session_int=session_int
)
# self.current_url = url # SimServer.receive 会返回新的 URL，应使用那个
# click 方法：模拟用户点击页面上的一个元素
def click(self, clickable_name, text_to_clickable):
"""Wrapper for `receive` handler for performing click action on current page"""
# 调用 SimServer 的 receive 方法，传递点击的元素名称
self.page_source, self.current_url, status = self.server.receive(
self.session_id,
current_url=self.current_url,
clickable_name=clickable_name,
text_to_clickable=text_to_clickable, # 传递可点击元素映射用于服务器端验证
)
return status # 返回服务器处理结果 (奖励和是否结束)
# search 方法：模拟用户在搜索框输入并提交搜索
def search(self, keywords):
"""Wrapper for `receive` handler for performing search action on current page"""
if isinstance(keywords, str):
keywords = keywords.split(" ") # 将关键词字符串按空格分割为列表
# 调用 SimServer 的 receive 方法，传递搜索关键词
self.page_source, self.current_url, status = self.server.receive(
self.session_id,
current_url=self.current_url,
keywords=keywords,
)
return status # 返回服务器处理结果
# 注册 Gym 环境
register(
id="WebAgentTextEnv-v0",
entry_point=(
"personalized_shopping.shared_libraries.web_agent_site.envs.web_agent_text_env:WebAgentTextEnv"
),
)1. WebAgentTextEnv 类：Gym 环境的实现
这是整个文件的核心，它继承自 gym.Env，遵循 OpenAI Gym 的环境接口规范。这使得该环境可以方便地与各种强化学习算法或其他基于 Gym 环境的 Agent 框架集成。
初始化 (__init__):
观测模式 (observation_mode): 定义了 Agent 从环境中获取的观测信息的格式。
'html': 返回原始的 HTML 页面源码。
'text': 返回从 HTML 中提取的、用 [SEP] 分隔的纯文本内容（通过 convert_html_to_text(simple=True) 实现）。
'text_rich': 返回带有结构化标记的文本内容，例如 [button] 文本 [button_]，以区分可点击元素和普通文本（通过 convert_html_to_text(simple=False) 实现）。
'url': 返回当前页面的 URL。
文件路径 (file_path): 指向包含产品数据的 JSON 文件。
服务器 (server):
可以传入一个已有的 SimServer 实例，或者在内部创建一个新的 SimServer 实例。
SimServer 负责模拟后端 Web 服务器的逻辑。
浏览器 (browser):
创建一个 SimBrowser 实例，它负责模拟浏览器的行为，如页面加载和点击。
会话管理 (session, session_prefix): 用于跟踪和区分不同的用户交互会话。
图像特征加载 (可选 get_image): 如果启用了图像功能，会加载预先计算的图像特征 (FEAT_CONV) 和对应的 ID 映射 (FEAT_IDS)。
交互历史 (prev_obs, prev_actions, num_prev_obs, num_prev_actions):
存储先前若干步的观测和动作。这对于需要理解上下文的 Agent 非常重要，可以将历史信息作为当前状态的一部分。
重置 (self.reset()): 在初始化结束时调用 reset 方法，准备好第一个会话。
step(self, action) 方法:
这是 Gym 环境的核心方法，用于执行 Agent 的一个动作并返回结果。
动作解析 (parse_action): 将输入的动作字符串（如 "search[keywords]" 或 "click[button_name]"）解析为动作名称和参数。
动作执行:
Search: 如果是搜索动作，且参数有效，则调用 self.browser.search(action_arg)。
Click: 如果是点击动作，且点击目标存在于当前页面的可点击元素列表 (self.text_to_clickable) 中，则调用 self.browser.click(action_arg, self.text_to_clickable)。
无效动作: 如果动作无效，则不执行任何操作，返回一个表示没有奖励和未结束的状态。
状态更新与返回:
获取新的观测 (ob = self.observation)。
将当前动作和观测添加到交互历史中。
根据配置的 num_prev_obs 和 num_prev_actions，将历史观测和动作与当前观测拼接成新的状态 state，并用 [SEP] 分隔。
返回 state (新的观测状态)、reward (奖励)、done (是否结束) 和 info (附加信息，当前版本为 None)。
reset(self, session=None, instruction_text=None) 方法:
重置环境到一个新的初始状态，通常在开始一个新会话或当前会话结束后调用。
会话 ID 管理:
如果提供了 session 参数，则使用该会话 ID；否则，随机生成一个新的会话 ID。
如果设置了 session_prefix，会给会话 ID 添加前缀。
初始化 URL: 构建初始页面的 URL（通常是搜索页面）。
浏览器导航: 调用 self.browser.get(init_url, ...) 来加载初始页面。
指令文本获取:
如果提供了 instruction_text 参数，则使用它。
否则，调用 self.get_instruction_text() 从当前页面（通常是新会话的初始页面）获取指令文本。这个指令文本通常是预设的购物目标。
清空历史: 清空 prev_obs 和 prev_actions，并将当前观测作为历史的第一个元素。
返回初始观测和 None (info)。
get_available_actions(self) 方法:
解析当前页面的 HTML，找出所有可点击的元素。
HTML 解析: 使用 BeautifulSoup 解析 self.browser.page_source。
元素查找:
查找搜索输入框 (id="search_input")。
查找所有类别为 btn 的按钮。
查找所有类别为 product-link 的产品链接。
查找所有 input[type="radio"] 的购买选项。
构建映射 (self.text_to_clickable): 将这些可点击元素的文本内容（小写）映射到它们对应的 BeautifulSoup 元素对象。这用于在 step 方法中验证点击动作的有效性。
返回一个字典，包含 has_search_bar (布尔值) 和 clickables (可点击元素文本列表)。
get_image(self) 方法 (可选功能):
如果启用了图像功能，此方法用于从当前产品页面提取图像特征。
解析 HTML 找到产品图片元素 (id="product-image")。
获取图片 URL，并在预加载的图像特征库 (self.feats 和 self.ids) 中查找对应的特征向量。
如果找到，返回特征向量；否则返回零向量。
get_instruction_text(self) 方法:
从当前页面的 HTML 中提取指令文本（通常是购物目标描述）。
解析 HTML 找到指令文本元素 (id="instruction-text")。
_parse_html(self, html=None) 方法:
一个辅助方法，用于将 HTML 字符串包装成 BeautifulSoup 对象，方便后续解析。
如果未提供 html 参数，则默认使用当前状态的 HTML (self.state["html"])。
observation (属性):
根据初始化时设置的 self.observation_mode，返回不同格式的观测信息。
调用 self.convert_html_to_text 将 HTML 转换为不同风格的文本表示。
state (属性):
返回一个包含环境完整信息的字典，包括当前 url、html 源码和 instruction_text。Agent 的实际观测 (self.observation) 通常是这个完整状态的子集或转换形式。
convert_html_to_text(self, html, simple=False) 方法:
将 HTML 源码转换为文本表示，这是实现不同 observation_mode 的关键。
使用 BeautifulSoup 提取所有可见文本内容（通过 tag_visible 过滤）。
simple=True (对应 'text' 模式):
将所有可见文本去除首尾空格后，用 [SEP] 连接起来。
simple=False (对应 'text_rich' 模式):
为不同类型的 HTML 元素（如按钮、选项、产品链接）添加特定的标记，如 [button] 文本 [button_] 或 [clicked button] 文本 [clicked button_]。
这有助于 Agent 更好地区分页面上的可交互元素和普通文本，并理解哪些元素已经被点击过。
render(self, mode="human") 和 close(self) 方法:
标准的 Gym 环境方法，但在这个实现中它们是空操作 (pass)。实际的渲染（如果需要）通常通过查看 Agent 保存的 HTML 工件或直接在模拟的 Web 界面（如果项目支持）中进行。
2. SimServer 类：轻量级 Web 服务器模拟器
SimServer 负责模拟 WebShop 应用的后端逻辑。它接收来自 SimBrowser 的“请求”（实际上是方法调用），处理这些请求，并返回相应的 HTML 页面内容和状态更新。
初始化 (__init__):
加载数据:
调用 load_products (来自 engine.py) 加载所有产品信息、产品字典 (ASIN -> 产品对象)、产品价格。
调用 init_search_engine (来自 engine.py) 初始化 Pyserini 搜索引擎。
调用 get_goals (来自 goal.py) 加载或生成购物目标。
参数配置:
filter_goals: 一个可选函数，用于根据特定标准筛选目标。
limit_goals: 限制可用的目标数量。
num_products: 控制加载和索引的产品数量。
human_goals: 是否加载人类标注的目标。
show_attrs: 是否在商品页面显示额外属性。
会话存储 (self.user_sessions): 一个字典，用于存储每个会话的状态信息。
性能计时: 包含 search_time 和 render_time 用于记录搜索和渲染的耗时（主要用于分析，不直接影响 Agent）。
assigned_instruction_text: 一个特殊的变量，允许外部（如 search.py 工具）临时覆盖当前会话的指令文本显示。这是一个有些 hacky 的实现，用于在搜索时动态更新页面上的指令。
模拟路由处理方法 (如 index, search_results, item_page, item_sub_page, done):
这些方法虽然使用了 Flask 风格的 @app.route 装饰器（在 SimServer 的上下文中），但它们实际上是被 receive 方法调用的内部逻辑处理单元。
每个方法对应 WebShop 的一个页面或一种状态。
它们负责：
更新对应会话 (session = self.user_sessions[session_id]) 的状态（如关键词、当前页码、选择的 ASIN、选项等）。
执行核心逻辑（如调用搜索引擎 get_top_n_product_from_keywords，获取特定商品信息 self.product_item_dict[session["asin"]]）。
构建新的 URL。
调用 map_action_to_html (来自 engine.py) 来渲染对应页面的 HTML。
在 done 方法中，调用 get_reward (来自 goal.py) 计算最终奖励。
返回生成的 html, url，有时还包括 reward。
receive(self, session_id, current_url, session_int=None, **kwargs) 方法:
这是 SimServer 的核心入口点，模拟了接收 HTTP 请求并返回响应的过程。
会话初始化与目标分配:
如果 session_id 是新的，会随机选择或根据 session_int 指定一个购物目标，并初始化会话的基本信息（如空关键词、无已选商品、动作计数器等）。
从目标中提取 instruction_text。
请求分发:
无 kwargs (初始页面): 调用 self.index 返回搜索首页。
包含 keywords: 调用 self.search_results 执行搜索并返回结果页。
包含 clickable_name:
"Buy Now" (END_BUTTON): 调用 self.done，计算奖励，标记会话结束。
"Back to Search" (BACK_TO_SEARCH): 递归调用 self.receive 返回搜索首页。
"Next >" (NEXT_PAGE) / "< Prev" (PREV_PAGE) 在搜索结果页: 调整页码，重新调用 self.receive (带着关键词和新页码) 来获取新的搜索结果页。
"< Prev" (PREV_PAGE) 在商品子页面: 调用 self.item_page 返回商品主详情页。
"< Prev" (PREV_PAGE) 在商品主详情页: 调用 self.search_results 返回之前的搜索结果页。
点击 "Description", "Features", "Reviews" (ACTION_TO_TEMPLATE中的键): 调用 self.item_sub_page 显示商品子信息页。
其他点击 (通常是商品链接或购买选项): 调用 self.item_page 更新商品选项或显示商品主详情页。
返回最终的 html, url, 和 status (包含奖励和是否结束的字典)。
get_page_name(self, url) 方法:
一个辅助方法，根据 URL 字符串判断当前属于哪个类型的页面（如 search_results, item_page）。
3. SimBrowser 类：模拟浏览器行为
SimBrowser 封装了与 SimServer 交互的逻辑，模拟了用户在浏览器中的操作。
初始化 (__init__):
接收一个 SimServer 实例作为参数。
初始化 current_url, page_source (当前页面 HTML), session_id。
get(self, url, session_id=None, session_int=None) 方法:
模拟浏览器加载一个新 URL。
更新 self.session_id。
调用 self.server.receive (不带额外参数，通常用于加载初始页面或通过 URL 直接导航)，获取页面 HTML 和新的 URL。
更新 self.page_source 和 self.current_url。
click(self, clickable_name, text_to_clickable) 方法:
模拟用户点击页面上的一个元素。
调用 self.server.receive，传入 clickable_name 和 text_to_clickable (当前页面的可点击元素映射，用于服务器端验证)。
更新 self.page_source, self.current_url，并返回服务器返回的 status (奖励和是否结束)。
search(self, keywords) 方法:
模拟用户在搜索框输入并提交搜索。
将关键词（如果是字符串，则按空格分割为列表）传递给 self.server.receive。
更新 self.page_source, self.current_url，并返回服务器返回的 status。
4. 辅助函数 tag_visible(element)
这是一个用于 BeautifulSoup 文本提取的辅助函数。
它判断一个 HTML 元素是否应该被视为“可见”文本。
会忽略 <style>, <script>, <head>, <title>, <meta> 标签以及 HTML 注释 (Comment) 中的内容。
总结 web_agent_text_env.py 的核心价值：
提供了一个可控、可复现的交互环境: 使得 Agent 可以在一个模拟的 WebShop 中进行学习和测试，而无需真实地部署一个复杂的 Web 应用和搜索引擎。
抽象了复杂的 Web 交互: 将底层的 HTML 解析、状态管理、页面渲染等复杂性封装起来，Agent 只需要关注高层次的动作（搜索、点击）和观测（文本表示的页面内容）。
支持多种观测模式: 允许研究者和开发者根据 Agent 的能力和需求，选择不同粒度的信息作为输入。text_rich 模式尤其有用，因为它保留了文本的结构信息。
集成了目标和奖励机制: 虽然目标生成和奖励计算的逻辑主要在 goal.py 和 engine.py 中，但 SimServer 将它们整合到会话流程中，为评估 Agent 性能提供了基础。
模块化设计: WebAgentTextEnv, SimServer, SimBrowser 各司其职，使得代码结构清晰，易于理解和扩展。
通过理解这些功能，您可以在博客中清晰地阐述 web_agent_text_env.py 如何为个性化购物 Agent 构建一个功能完备且易于操作的模拟世界。
init_env.py 功能详解
shared_libraries/init_env.py 在整个项目中扮演着一个看似简单但至关重要的“启动与配置中心”的角色。它的核心任务是创建、配置并初始化一个全局唯一的 WebShop 环境实例，供项目的所有其他部分（尤其是 Agent 的工具）共享使用。
# 导入 gym 库，它是创建和管理模拟环境的标准框架。
import gym
# --- 1. 注册自定义 Gym 环境 ---
# 向 Gym 框架注册我们自定义的 WebShop 环境。
# 这样，我们就可以通过 gym.make("WebAgentTextEnv-v0") 来创建这个环境的实例。
gym.envs.registration.register(
# id: 为自定义环境指定的唯一标识符。
id="WebAgentTextEnv-v0",
# entry_point: 指向实现该环境的 Python 类。
# 格式为 '模块路径:类名'。
# 这里指定了环境的实现类是 WebAgentTextEnv，位于 ...web_agent_text_env.py 文件中。
entry_point=(
"personalized_shopping.shared_libraries.web_agent_site.envs.web_agent_text_env:WebAgentTextEnv"
),
)
# --- 2. 定义标准化的环境初始化函数 ---
# 定义一个函数，用于封装创建和配置 WebShop 环境的过程。
def init_env(num_products):
"""
初始化并返回一个配置好的 WebAgentTextEnv 环境实例。
Args:
num_products (int): 要加载到环境中的产品数量。
Returns:
gym.Env: 一个配置好的 WebAgentTextEnv 环境实例。
"""
# 使用 gym.make 并传入环境 ID 来创建环境实例。
env = gym.make(
"WebAgentTextEnv-v0",
# observation_mode: 配置 Agent 从环境中获取的观测信息格式。
# 'text' 模式意味着 Agent 看到的是经过处理的、更易于 LLM 理解的纯文本内容。
observation_mode="text",
# num_products: 控制环境中加载的产品数量，直接影响内存消耗和初始化时间。
num_products=num_products,
)
return env
# --- 3. 创建并共享全局环境实例 ---
# 设置要加载到环境中的默认产品数量。
# 50,000 是一个在真实性和性能之间权衡的数值。
num_product_items = 50000
# 调用初始化函数，创建整个项目共享的全局环境实例。
# 这个 `webshop_env` 对象将被 Agent 的工具（如 click.py, search.py）导入和使用。
webshop_env = init_env(num_product_items)
# 在创建环境后，必须调用 reset() 方法来初始化第一个会话。
# 这会加载初始页面，并准备好第一个观测，使环境进入一个可交互的初始状态。
webshop_env.reset()
# 打印一条日志，确认环境初始化已完成，并告知加载的商品数量。
print(f"Finished initializing WebshopEnv with {num_product_items} items.")init_env.py 在整个个性化购物 Agent 项目中扮演着一个至关重要的初始化和全局共享的角色。它的功能虽然集中，但对项目的运行至关重要，可以概括为以下三点：
注册自定义 Gym 环境:
代码的第一部分 gym.envs.registration.register(...) 负责向 OpenAI Gym 框架注册一个自定义的环境。
id="WebAgentTextEnv-v0": 这是为自定义环境指定的唯一 ID。当代码中出现 gym.make("WebAgentTextEnv-v0") 时，Gym 框架就知道要去实例化哪个类。
entry_point: 这是关键部分，它告诉 Gym 框架，当需要创建 WebAgentTextEnv-v0 环境时，应该去哪个 Python 模块 (personalized_shopping.shared_libraries.web_agent_site.envs.web_agent_text_env) 寻找哪个类 (WebAgentTextEnv)。
重要性: 如果没有这一步注册，gym.make() 将无法找到并创建我们的自定义环境，整个项目将无法运行。它是在 Gym 的“生态系统”中为我们的环境“上户口”。
提供一个标准化的环境初始化函数 (init_env):
该文件定义了一个名为 init_env(num_products) 的函数。这个函数封装了创建和配置 WebAgentTextEnv 环境实例的具体过程。
封装细节: 它调用 gym.make("WebAgentTextEnv-v0", ...) 来创建环境实例，并传递了两个重要的配置参数：
observation_mode="text": 明确指定了 Agent 从环境中获取的观测信息是纯文本格式。这意味着 Agent 看到的是经过处理的、去除了 HTML 标签的页面内容，而不是原始的 HTML 源码。这对基于文本的 LLM 来说是更友好、更高效的输入。
num_products=num_products: 这个参数控制了模拟环境中加载和索引的产品数量。这是一个非常重要的性能和资源控制参数。加载的产品越多，环境的真实性越高，但同时也会消耗更多的内存和初始化时间。
目的: 通过提供这样一个函数，项目的其他部分（例如，未来可能的训练脚本或不同的 Agent 实现）可以以一种统一、简洁的方式来获取一个配置好的环境实例，而无需关心 gym.make 的具体参数细节。
创建并共享一个全局环境实例 (webshop_env):
这是该文件最核心的实际作用。在定义了 init_env 函数之后，代码立即调用它来创建一个全局的、可被项目内其他模块共享的环境实例，命名为 webshop_env。
num_product_items = 50000: 在调用前，设定了默认加载 50,000 个商品。这是一个在真实性和性能之间取得平衡的默认值。
webshop_env = init_env(num_product_items): 创建环境实例。
webshop_env.reset(): 创建后立即调用 reset() 方法。这一步是必需的，它会初始化环境的第一个会话，加载初始页面，并准备好第一个观测，确保环境处于一个可用的初始状态。
全局共享的重要性: 在这个项目中，Agent 的工具（如 click.py 和 search.py）需要直接操作环境。通过在 init_env.py 中创建一个全局的 webshop_env 实例，这些工具模块就可以简单地通过 from ..shared_libraries.init_env import webshop_env 来导入并使用同一个环境实例。这确保了 Agent 和它的工具操作的是同一个世界，维持了交互状态的一致性。如果每个工具都创建自己的环境实例，那么它们将工作在不同的、互不相干的模拟世界中，无法实现连贯的交互。
总结: init_env.py 文件是一个项目的启动配置文件和核心共享枢纽。它首先向 Gym 框架“声明”了自定义环境的存在，然后定义了一个标准的创建流程，并最终创建了一个全局的、预配置好的环境实例 webshop_env。这个全局实例是连接 Agent 工具和模拟环境的桥梁，是整个 Agent 能够与模拟 WebShop 进行交互的基础。
agent.py 功能详解
agent.py 可以说是整个个性化购物 Agent 项目的“大脑”定义文件或“蓝图”。它的核心作用是实例化和配置 Agent，将项目的各个关键部分——大型语言模型（LLM）、核心指令（Prompt）和可用工具（Tools）——组装在一起，形成一个完整、可运行的智能体。
# 从 Agent Development Kit (ADK) 导入核心的 Agent 类，用于构建智能体。
from google.adk.agents import Agent
# 导入 FunctionTool，用于将普通 Python 函数包装成 Agent 可调用的工具。
from google.adk.tools import FunctionTool
# 从 .tools 目录导入自定义的 search 和 click 函数。
# search: 用于在模拟的 WebShop 环境中搜索商品。
# click: 用于在模拟的 WebShop 环境中点击按钮或链接。
from .tools.search import search
from .tools.click import click
# 从 prompt.py 导入详细的、指导 Agent 行为的核心指令（System Prompt）。
# 这个指令告诉 Agent 它的角色、目标、以及如何一步步与用户和环境交互。
from .prompt import personalized_shopping_agent_instruction
# 导入 LiteLlm，这是一个灵活的模型包装器，用于连接非原生支持的 LLM 服务。
from google.adk.models.lite_llm import LiteLlm
# 实例化 Agent，定义其所有核心组件，并将其赋值给 root_agent。
# 这个 root_agent 对象就是 ADK 框架将要运行的智能体实例。
root_agent = Agent(
# 配置 Agent 的“大脑”：使用的大型语言模型 (LLM)。
# 这里使用 LiteLlm 包装器来连接到 Grok 模型服务。
model=LiteLlm(
model="xai/grok-3-beta", # 指定要使用的具体模型名称。
api_base="https://api.x.ai/v1", # 该模型服务的 API 端点地址。
api_key="xxxxxxxx" # 调用 API 所需的密钥。
),
# 为 Agent 指定一个唯一的名称，便于识别和管理。
name="personalized_shopping_agent",
# 为 Agent 设置核心指令（“系统提示”），指导其行为、个性和目标。
instruction=personalized_shopping_agent_instruction,
# 为 Agent 配备其可用的工具集（“手和眼”），使其能够与外部环境交互。
tools=[
# 将 search 函数包装成一个 Agent 可用的工具。
FunctionTool(
func=search,
),
# 将 click 函数包装成一个 Agent 可用的工具。
FunctionTool(
func=click,
),
],
)虽然agent.py代码量很少，但其重要性极高，因为它定义了 Agent 的“身份”和“能力”。
1. 代理（Agent）的实例化
文件最核心的部分是创建了一个 Agent 类的实例，命名为 root_agent。在 Google Agent Development Kit (ADK) 框架中，Agent 类是构建智能体的基础。这个实例就是我们最终与之交互的那个“个性化购物 Agent”。
2. 模型集成（The Brain）
Agent 的智能和推理能力来源于其背后的大型语言模型 (LLM)。
model 参数: 这个参数指定了 Agent 使用哪个 LLM 作为其“大脑”。
LiteLlm: 代码中使用了 LiteLlm 包装器。这是一个非常灵活的工具，它允许 ADK 框架连接到各种非原生的 LLM 服务。这极大地增强了项目的可扩展性。
具体模型配置:
model="xai/grok-3-beta": 这里作者指定了使用 Grok-3 Beta 模型。
api_base="https://api.x.ai/v1": 这是 Grok 模型服务的 API 端点地址。
api_key="...": 这是调用该模型服务所需的身份验证密钥。
通过这个配置，Agent 的所有推理任务（如理解用户意图、决定下一步行动、选择调用哪个工具、生成自然语言回复等）都会被发送到指定的 Grok 模型进行处理。
3. 核心指令（The Soul）
如果说 LLM 是 Agent 的大脑，那么核心指令（Prompt）就是它的“灵魂”或“行为准则”。
instruction 参数: 这个参数接收一个长字符串，作为 LLM 的系统提示（System Prompt）。这个提示至关重要，它为 Agent 设定了角色、目标和详细的行动指南。
personalized_shopping_agent_instruction: 代码从 prompt.py 文件中导入了这个变量。prompt.py 中包含了详细的交互流程，例如：
如何处理用户的初始请求。
如何使用 search 工具。
如何探索产品详情页（包括主动点击“Description”、“Features”等）。
如何处理购买选项（颜色、尺寸）。
如何与用户确认购买。
以及最重要的，如何正确地使用导航按钮如 < Prev 和 Back to Search。
这个精心设计的指令确保了 Agent 的行为是可控且符合预期的，而不是漫无目的地与环境交互。
4. 工具集（The Hands and Eyes）
为了能与外部世界（在这个项目中是模拟的 WebShop 环境）交互，Agent 需要工具。
tools 参数: 这个参数接收一个列表，列表中是 Agent 可以使用的所有工具。
FunctionTool: 这是 ADK 框架提供的一个包装器，它可以将一个普通的 Python 函数（如我们项目中的 search 和 click 函数）转换成 LLM 可以理解和调用的工具。
工具导入: 代码从 .tools.search 和 .tools.click 中导入了 search 和 click 函数。
工作流程:
当 LLM 在其推理过程中认为需要搜索或点击时，它不会直接执行 Python 代码。
相反，它会生成一个结构化的输出，表明它想要调用某个工具并提供相应的参数（例如，{"tool_call": "search", "parameters": {"keywords": "summer dress"}}）。
ADK 框架会捕获这个输出，找到名为 search 的 FunctionTool，并用提供的参数 {"keywords": "summer dress"} 来执行底层的 Python search 函数。
search 函数执行后会返回结果（例如，搜索页面的文本内容），这个结果会再次被送回给 LLM，作为它下一步决策的依据。
通过这种方式，Agent 被赋予了“看”（通过工具获取信息）和“做”（通过工具改变环境状态）的能力。
总结：agent.py 文件通过一个简洁的 Agent 实例化过程，将一个强大的语言模型、一套详细的行为指令和一组功能明确的工具绑定在一起，最终定义出了一个功能完备、目标明确的个性化购物 Agent。这个被定义的 root_agent 对象是整个应用程序的入口点，可以被 ADK 的命令行工具 (adk run) 或 Web 界面 (adk web) 加载和运行。
prompt.py 功能详解
prompt.py 在整个个性化购物 Agent 项目中扮演着至关重要的角色，可以将其视为 Agent 的“灵魂”或“行为手册”。它的核心功能是定义一个详细、结构化的系统提示（System Prompt），这个提示被注入到大型语言模型（LLM）中，以指导和约束 Agent 的行为。
personalized_shopping_agent_instruction = """You are a webshop agent, your job is to help the user find the product they are looking for, and guide them through the purchase process in a step-by-step, interactive manner.
**Interaction Flow:**
1. **Initial Inquiry:**
* Begin by asking the user what product they are looking for if they didn't provide it directly.
* If they upload an image, analyze what's in the image and use that as the reference product.
2. **Search Phase:**
* Use the "search" tool to find relevant products based on the user's request.
* Present the search results to the user, highlighting key information and available product options.
* Ask the user which product they would like to explore further.
3. **Product Exploration:**
* Once the user selects a product, automatically gather and summarize all available information from the "Description," "Features," and "Reviews" sections.
* You can do this by clicking any of the "Description," "Features," or "Reviews" buttons, navigate to the respective section and gather the information. After reviewing one section, return to the information page by clicking the "< Prev" button, then repeat for the remaining sections.
* Avoid prompting the user to review each section individually; instead, summarize the information from all three sections proactively.
* If the product is not a good fit for the user, inform the user, and ask if they would like to search for other products (provide recommendations).
* If the user wishes to proceed to search again, use the "Back to Search" button.
* Important: When you are done with product exploration, remeber to click the "< Prev" button to go back to the product page where all the buying options (colors and sizes) are available.
4. **Purchase Confirmation:**
* Click the "< Prev" button to go back to the product page where all the buying options (colors and sizes) are available, if you are not on that page now.
* Before proceeding with the "Buy Now" action, click on the right size and color options (if available on the current page) based on the user's preference.
* Ask the user for confirmation to proceed with the purchase.
* If the user confirms, click the "Buy Now" button.
* If the user does not confirm, ask the user what they wish to do next.
5. **Finalization:**
* After the "Buy Now" button is clicked, inform the user that the purchase is being processed.
* If any errors occur, inform the user and ask how they would like to proceed.
**Key Guidelines:**
* **Slow and Steady:**
* Engage with the user when necessary, seeking their input and confirmation.
* **User Interaction:**
* Prioritize clear and concise communication with the user.
* Ask clarifying questions to ensure you understand their needs.
* Provide regular updates and seek feedback throughout the process.
* **Button Handling:**
* **Note 1:** Clikable buttons after search look like "Back to Search", "Next >", "B09P5CRVQ6", "< Prev", "Descriptions", "Features", "Reviews" etc. All the buying options such as color and size are also clickable.
* **Note 2:** Be extremely careful here, you must ONLY click on the buttons that are visible in the CURRENT webpage. If you want to click a button that is from the previous webpage, you should use the "< Prev" button to go back to the previous webpage.
* **Note 3:** If you wish to search and there is no "Search" button, click the "Back to Search" button instead."""为了便于理解，我们将其翻译成中文：
personalized_shopping_agent_instruction = """你是一个网店代理 (webshop agent)，你的工作是帮助用户找到他们正在寻找的产品，并以分步、互动的方式引导他们完成购买过程。
**交互流程：**
1. **初始问询：**
* 如果用户没有直接提供他们想要的产品，首先询问他们正在寻找什么。
* 如果他们上传了一张图片，分析图片中的内容，并将其作为参考产品。
2. **搜索阶段：**
* 使用 "search" 工具根据用户的请求查找相关产品。
* 向用户展示搜索结果，突出关键信息和可用的产品选项。
* 询问用户希望进一步探索哪个产品。
3. **产品探索：**
* 一旦用户选择了一个产品，自动地从“Description”（描述）、“Features”（特性）和“Reviews”（评论）部分收集并总结所有可用信息。
* 你可以通过点击任何“Description”、“Features”或“Reviews”按钮，导航到相应的部分并收集信息。在查看完一个部分后，通过点击“< Prev”（上一页）按钮返回到信息页面，然后对剩下的部分重复此操作。
* 避免提示用户逐一查看每个部分；相反，要主动地总结所有三个部分的信息。
* 如果产品不适合用户，告知用户，并询问他们是否希望搜索其他产品（提供建议）。
* 如果用户希望再次进行搜索，请使用“Back to Search”（返回搜索）按钮。
* 重要提示：当你完成产品探索后，记得点击“< Prev”（上一页）按钮，返回到有所有购买选项（颜色和尺寸）的产品页面。
4. **购买确认：**
* 如果你现在不在有所有购买选项（颜色和尺寸）的产品页面，请点击“< Prev”（上一页）按钮返回到该页面。
* 在执行“Buy Now”（立即购买）操作之前，根据用户的偏好，在当前页面上点击正确的尺寸和颜色选项（如果可用）。
* 请求用户确认是否继续购买。
* 如果用户确认，点击“Buy Now”（立即购买）按钮。
* 如果用户不确认，询问用户接下来希望做什么。
5. **最终完成：**
* 在点击“Buy Now”（立即购买）按钮后，告知用户购买正在处理中。
* 如果发生任何错误，告知用户并询问他们希望如何继续。
**关键指南：**
* **稳步推进 (Slow and Steady)：**
* 在必要时与用户互动，征求他们的输入和确认。
* **用户交互 (User Interaction)：**
* 优先与用户进行清晰简洁的沟通。
* 提出澄清性问题，以确保你理解他们的需求。
* 在整个过程中提供定期的更新并寻求反馈。
* **按钮处理 (Button Handling)：**
* **注意 1：** 搜索后可点击的按钮看起来像 "Back to Search"、"Next >"、"B09P5CRVQ6"、"< Prev"、"Descriptions"、"Features"、"Reviews" 等。所有的购买选项，如颜色和尺寸，也是可点击的。
* **注意 2：** 这里要格外小心，你必须只点击当前网页中可见的按钮。如果你想点击一个来自前一个网页的按钮，你应该使用“< Prev”（上一页）按钮返回到前一个网页。
* **注意 3：** 如果你希望进行搜索但没有“Search”按钮，请点击“Back to Search”（返回搜索）按钮作为替代。"""LLM 本身是一个通用的语言模型，但通过这个精心设计的提示，我们可以将其“塑造”成一个专业的、目标明确的个性化购物助手。这个提示词文件完成了以下几个关键任务：
角色定义 (Role-Playing):
第一句话就为 Agent 设定了明确的角色：“你是一个网店代理 (webshop agent)”。这让 LLM 立即进入特定场景，而不是作为一个通用的问答机器人。
核心目标设定 (Goal Setting):
明确了 Agent 的核心任务：“帮助用户找到他们正在寻找的产品，并以分步、互动的方式引导他们完成购买过程。” 这为 Agent 的所有行动提供了最终的导向。
行为流程规范化 (Process Standardization):
交互流程 (Interaction Flow) 部分是该提示的核心。它将复杂的购物任务分解成五个清晰、有序的阶段：
初始问询 (Initial Inquiry): 如何开始对话，如何处理用户直接或间接（如通过图片）提供的需求。
搜索阶段 (Search Phase): 如何使用 search 工具，如何呈现结果，以及如何引导用户选择。
产品探索 (Product Exploration): 这是一个非常关键的环节，提示要求 Agent 主动地、自动化地去收集信息（点击“Description”、“Features”、“Reviews”），而不是被动地等待用户指令。这体现了 Agent 的智能性和前瞻性。它还规定了在探索后要记得返回到有购买选项的页面。
购买确认 (Purchase Confirmation): 规定了在购买前必须先处理购买选项（如颜色、尺寸），并与用户进行最终确认，体现了流程的严谨性。
最终完成 (Finalization): 规定了购买后的告知流程和错误处理。
这种流程化的规定将一个开放式的任务变成了一个结构化的工作流，大大提高了 Agent 行为的可靠性和可预测性。
关键指南和约束 (Key Guidelines & Constraints):
交互风格 (Slow and Steady): 提示 Agent 采取一种谨慎、需要时与用户确认的交互风格，避免鲁莽行动。
用户交互 (User Interaction): 强调了清晰沟通、提问澄清和寻求反馈的重要性。
按钮处理 (Button Handling): 这是技术性最强、也最容易出错的部分。提示通过三个“注意”点，对 Agent 的 click 工具使用进行了严格约束：
Note 1: 明确告知 Agent 页面上可能出现的按钮类型，帮助 LLM 更好地“理解”页面内容。
Note 2 (极其重要): 强调了只能点击当前页面可见的按钮。这是防止 Agent 产生“幻觉”（即试图点击已经不存在于当前页面的按钮）的关键约束。它还指导 Agent 如何使用 < Prev 按钮进行导航。
Note 3: 提供了在没有“Search”按钮时，使用“Back to Search”作为替代方案的策略。
总结：prompt.py 的功能远不止是简单的文本描述。它是一个精心设计的“软件规格说明书”，只不过它的“读者”是 LLM。通过角色扮演、目标设定、流程规范和行为约束，它将一个通用的 LLM 转化成一个专业的、可靠的、懂得如何使用工具来完成复杂任务的智能 Agent。这个文件的质量直接决定了整个 Agent 系统的上限和可靠性。
tools/click.py() 功能详解
click.py 文件定义了 Agent 在模拟 WebShop 环境中执行“点击”操作的核心逻辑。这个文件虽然简单，但它是 Agent 与环境进行交互、改变环境状态（即页面导航和选项选择）的关键“手臂”。
# 导入 ADK 的 ToolContext，用于访问工具执行时的上下文信息（如保存工件）。
from google.adk.tools import ToolContext
# 导入 genai 的 types，主要用于创建 Part 对象来保存工件。
from google.genai import types
# 从共享库中导入全局初始化的 WebShop 环境实例。
# 所有工具都通过操作这个全局实例来与模拟环境交互。
from ..shared_libraries.init_env import webshop_env
# 定义一个异步函数 `click`，它将作为 Agent 可用的工具。
async def click(button_name: str, tool_context: ToolContext) -> str:
"""
模拟在网页上点击一个具有给定名称的按钮或链接。
Args:
button_name(str): 要点击的按钮或链接的文本名称。
例如 "Next >", "B09P5CRVQ6", "Description", "red", "Buy Now" 等。
tool_context(ToolContext): ADK 框架传入的函数上下文，用于访问框架功能。
Returns:
str: 点击按钮后，新加载页面的文本观测信息。
"""
# 初始化一个字典来存储从环境中获取的状态信息。
status = {"reward": None, "done": False}
# 将点击意图构造成环境 `step` 方法可以理解的标准化动作字符串。
# 例如，如果 button_name 是 "Buy Now"，action_string 会是 "click[Buy Now]"。
action_string = f"click[{button_name}]"
# 执行动作：将动作字符串传递给模拟环境，并接收返回的状态。
# `webshop_env.step` 会模拟点击操作，并更新环境状态（如页面、奖励等）。
# _ (下划线) 用于接收我们在此处不关心的第一个返回值（新的状态/观测）。
_, status["reward"], status["done"], _ = webshop_env.step(action_string)
# 在环境状态更新后，获取新的观测信息。
# `webshop_env.observation` 包含了新页面的内容（格式由环境配置决定）。
ob = webshop_env.observation
# 对观测信息进行后处理：找到 "Back to Search" 出现的位置。
index = ob.find("Back to Search")
if index >= 0:
# 如果找到了，截取从该位置开始的后续所有内容。
# 这是一种简化观测的策略，旨在减少无关信息，让 LLM 更关注页面主体。
ob = ob[index:]
# --- 调试日志 ---
# 打印分隔线，便于在控制台清晰地查看每次点击的结果。
print("#" * 50)
print("Click result:")
# 打印点击后获得的状态（奖励和是否结束）。
print(f"status: {status}")
# 打印处理后的观测文本，这是将要返回给 LLM 的信息。
print(f"observation: {ob}")
print("#" * 50)
# --- 调试日志结束 ---
# 针对特定按钮的特殊处理逻辑。
if button_name == "Back to Search":
# 当点击 "Back to Search" 时，修改服务器的一个特殊变量。
# 这可能会改变下个页面上显示的指令文本，是一种定制化的行为。
webshop_env.server.assigned_instruction_text = "Back to Search"
# 在用户界面 (UI) 中显示工件 (Artifact)。
try:
# 使用 tool_context 将点击后生成的完整 HTML 页面保存为一个工件。
# 这对于调试非常有用，可以直观地查看 Agent 当时“看到”的页面。
await tool_context.save_artifact(
"html", # 工件的名称/类型
types.Part.from_uri(
# `webshop_env.state["html"]` 包含了原始的 HTML 源码。
file_uri=webshop_env.state["html"],
mime_type="text/html" # 指定工件的 MIME 类型，以便 UI 正确渲染。
),
)
except ValueError as e:
# 如果保存工件时出错（例如，在某些非 UI 环境下运行），则打印错误信息。
print(f"Error saving artifact: {e}")
# 返回处理后的观测信息给 Agent (LLM)，作为其下一步决策的依据。
return ob它的功能可以分解为以下几个部分：
定义为异步工具函数 (async def click(...)):
该函数被定义为 async，这意味着它被设计为在异步环境中运行。在现代的 Agent 和 Web 框架中，异步编程是常见的，可以提高 I/O 操作（如等待模型响应、文件读写等）的效率。
它接收两个关键参数：
button_name: str: 这是一个字符串，代表 Agent 想要点击的按钮或链接的文本内容。例如，"Next >", "B09P5CRVQ6", "Description", "red", "Buy Now" 等。
tool_context: ToolContext: 这是由 ADK (Agent Development Kit) 框架传入的上下文对象。它非常重要，因为它提供了访问框架特定功能（如保存工件）的途径。
构造动作字符串 (action_string = f"click[{button_name}]"):
这是将 Agent 的意图（点击某个按钮）转换为模拟环境 WebAgentTextEnv 能理解的标准化格式的关键一步。
WebAgentTextEnv 的 step 方法被设计为接收特定格式的字符串，如 click[...] 或 search[...]。
这一行代码创建了这样一个标准格式的动作字符串。
与环境交互 (webshop_env.step(action_string)):
这是整个函数的核心。它调用了全局初始化的 webshop_env 对象的 step 方法。
webshop_env 是 WebAgentTextEnv 的一个实例，它封装了整个模拟环境。
当 step 方法接收到 click[...] 动作时，它会触发 SimBrowser 和 SimServer 的一系列内部调用，模拟点击操作：
SimServer 会根据 button_name 判断应该导航到哪个新页面（搜索结果页、商品详情页、子信息页、完成页等），或者更新当前页面的状态（如选择了某个颜色）。
然后，SimServer 会渲染新的 HTML 页面。
step 方法最终会返回新页面的观测结果、奖励、是否结束的标志等信息。
代码 _, status["reward"], status["done"], _ = webshop_env.step(action_string) 负责接收并存储这些返回的状态。
获取并处理观测结果 (ob = webshop_env.observation):
在 step 方法执行完毕后，webshop_env 的内部状态已经更新，webshop_env.observation 属性现在会返回新页面的内容（根据配置，可能是 html、text 或 text_rich 格式）。
代码 index = ob.find("Back to Search") 和 ob = ob[index:] 是一个简单的后处理步骤。它的目的是截断观测信息，只保留从 "Back to Search" 按钮开始的部分。这可能是一种简化观测、减少不必要信息（如页面头部、重复的指令文本等）的策略，以减少传递给 LLM 的 token 数量，让模型更专注于页面主体内容。
日志记录 (print(...)):
在开发和调试过程中，打印详细的日志非常重要。这里打印了点击操作后的状态（奖励、是否结束）和处理后的观测结果，帮助开发者理解 Agent 的每一步交互是否符合预期。
特殊逻辑处理 (if button_name == "Back to Search"):
这里有一个针对特定按钮的硬编码逻辑。当 Agent 点击 "Back to Search" 时，代码会设置 webshop_env.server.assigned_instruction_text = "Back to Search"。
assigned_instruction_text 是 SimServer 中的一个特殊变量，它会临时覆盖页面上显示的指令文本。这可能是为了在返回搜索页面时，让页面上的指令文本显示为 "Back to Search"，以提醒 Agent 或用户当前的状态。这是一个特定于该项目实现的细节。
保存工件 (await tool_context.save_artifact(...)):
这是一个非常实用的功能，由 ADK 框架提供。
save_artifact 方法可以将任意数据作为“工件”与工具的本次调用关联起来。
在这里，它将点击后生成的完整 HTML 页面 (webshop_env.state["html"]) 保存为一个名为 "html" 的工件。
这样做的好处是，在调试或审查 Agent 的行为轨迹时，开发者不仅能看到 Agent 接收到的文本观测，还能直接查看当时 Agent “看到”的完整、可视化的 HTML 页面，极大地提高了可追溯性和调试效率。
总结：click.py 文件定义了一个作为 Agent “手臂”的工具。它将 LLM 的“点击”意图转换为环境可以理解的动作，执行这个动作，接收环境的反馈，对反馈进行初步处理，并利用 ADK 框架的功能记录详细的交互快照（HTML 工件），最终将处理后的观测结果返回给 LLM，以供其进行下一步决策。
tools/search.py 功能详解
search.py 文件为 Agent 定义了至关重要的“搜索”能力。在购物场景中，搜索是用户与电商平台交互的起点，也是 Agent 开始为用户寻找商品的第一步。这个文件定义了一个 search 工具，让 Agent 能够利用 WebShop 环境内建的搜索引擎来查找商品。
# 导入 ADK 的 ToolContext，用于访问工具执行时的上下文信息（如保存工件）。
from google.adk.tools import ToolContext
# 导入 genai 的 types，主要用于创建 Part 对象来保存工件。
from google.genai import types
# 从共享库中导入全局初始化的 WebShop 环境实例。
# 所有工具都通过操作这个全局实例来与模拟环境交互。
from ..shared_libraries.init_env import webshop_env
# 定义一个异步函数 `search`，它将作为 Agent 可用的工具。
async def search(keywords: str, tool_context: ToolContext) -> str:
"""
在模拟的 WebShop 环境中根据给定的关键词搜索商品。
Args:
keywords(str): 用于搜索商品的关键词，例如 "summer dress"。
tool_context(ToolContext): ADK 框架传入的函数上下文，用于访问框架功能。
Returns:
str: 搜索结果页面的文本观测信息。
"""
# 初始化一个字典来存储从环境中获取的状态信息。
status = {"reward": None, "done": False}
# 将搜索意图构造成环境 `step` 方法可以理解的标准化动作字符串。
# 例如，如果 keywords 是 "sunglasses"，action_string 会是 "search[sunglasses]"。
action_string = f"search[{keywords}]"
# --- 定制化页面指令 ---
# 在执行搜索前，动态地修改服务器端的一个特殊变量。
# 这会使得新加载的搜索结果页面上显示的指令文本反映当前的搜索任务。
# 例如，显示 "Find me sunglasses."，而不是通用的初始购物目标。
webshop_env.server.assigned_instruction_text = f"Find me {keywords}."
print(f"env instruction_text: {webshop_env.instruction_text}")
# 执行动作：将动作字符串传递给模拟环境，并接收返回的状态。
# `webshop_env.step` 会驱动搜索引擎进行查询，并渲染出搜索结果页面。
_, status["reward"], status["done"], _ = webshop_env.step(action_string)
# 在环境状态更新后，获取新的观测信息（即搜索结果页面的内容）。
ob = webshop_env.observation
# 对观测信息进行后处理，截取从 "Back to Search" 开始的部分。
# 这是一种简化观测、减少无关信息的策略，让 LLM 更关注页面主体。
index = ob.find("Back to Search")
if index >= 0:
ob = ob[index:]
# --- 调试日志 ---
print("#" * 50)
print("Search result:")
print(f"status: {status}") # 打印搜索后的状态（奖励通常为 None 或 0）。
print(f"observation: {ob}") # 打印处理后、将要返回给 LLM 的观测文本。
print("#" * 50)
# --- 调试日志结束 ---
# 在用户界面 (UI) 中显示工件 (Artifact)。
try:
# 使用 tool_context 将搜索后生成的完整 HTML 页面保存为一个工件。
# 这对于调试和回溯 Agent 的行为轨迹非常有用。
await tool_context.save_artifact(
"html", # 工件的名称/类型
types.Part.from_uri(
# `webshop_env.state["html"]` 包含了原始的 HTML 源码。
file_uri=webshop_env.state["html"],
mime_type="text/html" # 指定 MIME 类型以便 UI 正确渲染。
),
)
except ValueError as e:
# 如果保存工件时出错，打印错误信息。
print(f"Error saving artifact: {e}")
# 返回处理后的观测信息给 Agent (LLM)，作为其下一步决策的依据。
return ob它的核心功能可以分解为以下几个部分：
定义为异步工具函数 (async def search(...)):
与 click.py 类似，search 函数也被定义为 async，以适应异步编程框架。
它接收两个参数：
keywords: str: 一个字符串，代表用户想要搜索的关键词，例如 "summer dress", "blue running shoes" 等。这是 Agent 从用户对话或自己的推理中提取出的核心信息。
tool_context: ToolContext: 由 ADK 框架传入的上下文对象，用于访问框架功能，如保存工件。
构造动作字符串 (action_string = f"search[{keywords}]"):
这是将 Agent 的搜索意图转换为模拟环境 WebAgentTextEnv 能理解的标准化格式。
当 WebAgentTextEnv 的 step 方法接收到 search[...] 格式的动作时，它会调用 SimBrowser 的 search 方法，后者再触发 SimServer 的搜索逻辑。
定制化页面指令 (webshop_env.server.assigned_instruction_text = ...):
这是一个非常具体且重要的实现细节。在执行搜索之前，代码会动态地修改服务器端的一个变量 assigned_instruction_text。
这个变量会临时覆盖掉从购物目标（Goal）中来的原始指令。例如，如果用户搜索“太阳镜”，那么新加载的搜索结果页面顶部显示的指令文本就会变成“Find me sunglasses.”。
作用: 这样做可以提供更强的上下文感知。当 Agent（或用户）看到搜索结果页面时，页面上的指令明确地反映了当前的搜索任务，而不是一个通用的、可能已经不再相关的初始购物目标。这有助于 Agent 保持对当前任务的专注。
与环境交互 (webshop_env.step(action_string)):
这是函数的核心执行部分。它调用全局 webshop_env 实例的 step 方法。
step 方法会驱动 SimServer 调用 Pyserini 搜索引擎，根据 keywords 检索商品。
SimServer 拿到搜索结果后，会使用 results_page.html 模板渲染出搜索结果页面。
step 方法返回新页面的观测结果、奖励（搜索动作本身通常没有奖励，所以为 None 或 0）、是否结束等状态。
代码 _, status["reward"], status["done"], _ = webshop_env.step(action_string) 负责接收并存储这些返回的状态。
获取并处理观测结果 (ob = webshop_env.observation):
在 step 方法执行完毕后，webshop_env.observation 返回了新生成的搜索结果页面的内容。
与 click.py 类似，代码 index = ob.find("Back to Search") 和 ob = ob[index:] 对观测结果进行了截断处理。这同样是为了简化信息，移除页面头部等对 Agent 决策可能不那么重要的部分，减少 token 消耗。
日志记录 (print(...)):
打印详细的搜索结果日志，包括状态和处理后的观测文本，这对于调试和理解 Agent 的行为至关重要。
保存工件 (await tool_context.save_artifact(...)):
利用 ADK 框架的 save_artifact 功能，将搜索后生成的完整 HTML 搜索结果页面 (webshop_env.state["html"]) 保存为名为 "html" 的工件。
这使得开发者在回溯 Agent 行为时，可以直观地看到 Agent 执行搜索后所面对的完整页面，包括商品图片、标题、价格等，而不仅仅是处理过的文本。这对于分析 Agent 为何做出后续的点击决策非常有帮助。
总结：search.py 文件为 Agent 定义了一个关键的“眼睛”和“探索工具”。它将 LLM 的抽象搜索意图（通过关键词表达）转化为对模拟搜索引擎的具体调用。它不仅仅是简单地执行搜索，还通过动态修改页面指令来增强上下文感知，并通过保存 HTML 工件来提供强大的可追溯性。这个工具是 Agent 开始解决用户购物任务、获取初始信息集的入口。
convert_product_file_format.py，run_indexing.sh 功能详解
shared_libraries/search_engine/convert_product_file_format.py 和 shared_libraries/search_engine/run_indexing.sh 这两个文件是构建项目核心能力——产品搜索引擎——的关键步骤，它们协同工作，将原始的产品数据转换成可被高效检索的索引。
convert_product_file_format.py 的核心功能是数据预处理和格式转换。它扮演着一个数据管道（Data Pipeline）中 ETL (Extract, Transform, Load) 过程的“T”（Transform）角色。
具体来说，它的任务是：
加载原始产品数据: 从 items_shuffle.json 这个大型 JSON 文件中加载所有产品信息。这个原始文件的结构可能很复杂，包含了 Agent 可能不需要的所有原始字段。
提取和转换关键信息: 对于每个产品，脚本会提取出对搜索最关键的文本信息，并将它们整合成一个单一的、长字符串。这个过程包括：
提取基本信息: 获取产品的标题（Title）、描述（Description）、和要点（BulletPoints，通常是产品特性的简短列表）。
处理和拼接选项: 产品可能有多种购买选项，如颜色、尺寸、款式等（存储在 options 字段）。脚本会遍历这些选项，将它们转换成人类可读的文本格式（例如，"color: red, size: medium"），并拼接到主文本中。这使得用户可以直接搜索产品的特定选项（比如 "red dress"）。
格式化为搜索引擎兼容的格式: 脚本将每个产品处理后的信息转换成一个特定的 JSON 字典结构，这个结构是 Pyserini 搜索引擎所要求的。每个 JSON 对象包含：
id: 产品的唯一标识符，通常是 ASIN。这用于在搜索结果中唯一地识别产品。
contents: 一个包含所有关键文本信息的长字符串。搜索引擎将主要对这个字段进行索引和搜索。
product: 原始的、完整的商品信息字典。这使得在搜索到文档后，可以方便地从索引中取回完整的商品数据，而不仅仅是ID。
生成不同规模的数据集: 脚本最终会生成多个 .jsonl (JSON Lines) 文件，每个文件包含不同数量的产品（100, 1k, 10k, 50k）。这种分层的数据集设计非常实用，它允许：
快速开发和调试: 在开发阶段，可以使用小规模数据集（如 100 或 1k）快速地构建和测试索引，节省大量时间。
性能测试: 可以测试不同数据规模下搜索引擎的性能。
灵活部署: 根据部署环境的资源限制（内存、磁盘空间），可以选择加载相应规模的数据集和索引。
代码解析 (convert_product_file_format.py)
# 导入 json 库用于处理 JSON 数据，sys 用于修改 Python 路径，tqdm 用于显示进度条。
import json
import sys
from tqdm import tqdm
# 将父目录（项目根目录）添加到 Python 路径中，
# 这样就可以导入位于 web_agent_site 目录下的模块。
sys.path.insert(0, "../")
# 从项目的引擎模块中导入 load_products 函数。
from web_agent_site.engine.engine import load_products
# --- 1. 加载原始产品数据 ---
# 调用 load_products 函数加载 ../data/items_shuffle.json 文件中的所有产品。
# `*_` 表示我们只关心第一个返回值 all_products，忽略其他返回值。
all_products, *_ = load_products(filepath="../data/items_shuffle.json")
# --- 2. 遍历和处理每个产品 ---
docs = [] # 初始化一个列表，用于存储处理后的产品文档。
# 使用 tqdm 包装 all_products，可以在命令行中显示处理进度条。
for p in tqdm(all_products, total=len(all_products)):
# -- 2a. 处理和拼接选项文本 --
option_texts = [] # 存储当前产品所有选项的文本表示。
options = p.get("options", {}) # 获取产品的选项，如果不存在则返回空字典。
for option_name, option_contents in options.items():
# 将选项内容（如 ['red', 'blue']）用逗号连接成一个字符串（如 "red, blue"）。
option_contents_text = ", ".join(option_contents)
# 格式化成 "key: value" 的形式，如 "color: red, blue"。
option_texts.append(f"{option_name}: {option_contents_text}")
# 将所有选项文本用 ", and " 连接起来，形成一个完整的选项描述字符串。
option_text = ", and ".join(option_texts)
# -- 2b. 格式化为搜索引擎兼容的 JSON 对象 --
doc = dict()
doc["id"] = p["asin"] # 设置文档 ID 为产品的 ASIN。
# 将产品的标题、描述、第一个要点和处理后的选项文本拼接成一个大的 `contents` 字符串。
# 这个字符串是搜索引擎索引的核心内容。
# 使用 .lower() 将所有文本转为小写，以实现不区分大小写的搜索。
doc["contents"] = " ".join(
[
p["Title"],
p["Description"],
p["BulletPoints"][0] if p.get("BulletPoints") else "", # 确保 BulletPoints 存在
option_text,
]
).lower()
doc["product"] = p # 将完整的原始产品信息也存储在文档中。
docs.append(doc) # 将处理好的文档添加到列表中。
# --- 3. 生成不同规模的数据集文件 ---
# 将处理好的文档写入不同规模的 .jsonl 文件中。
# 'w+' 模式表示写入文件，如果文件不存在则创建。
# 使用切片操作 `docs[:100]` 来获取指定数量的文档。
# 每个 JSON 对象写入一行，符合 JSON Lines 格式。
with open("./resources_100/documents.jsonl", "w+") as f:
for doc in docs[:100]:
f.write(json.dumps(doc) + "\n")
with open("./resources_1k/documents.jsonl", "w+") as f:
for doc in docs[:1000]:
f.write(json.dumps(doc) + "\n")
with open("./resources_10k/documents.jsonl", "w+") as f:
for doc in docs[:10000]:
f.write(json.dumps(doc) + "\n")
with open("./resources_50k/documents.jsonl", "w+") as f:
for doc in docs[:50000]:
f.write(json.dumps(doc) + "\n")run_indexing.sh 是一个 Shell 脚本，它的功能是自动化地调用 Pyserini 搜索引擎库来为上一步生成的数据文件构建索引。索引是搜索引擎实现快速检索的基础，它类似于一本书的目录，可以让你快速定位到包含特定词语的页面（文档），而无需从头到尾阅读整本书。
这个脚本的核心作用是：
自动化流程: 它将为不同规模的数据集（100, 1k, 10k, 50k）构建索引的命令集总在一起，用户只需运行这一个脚本，就可以完成所有索引的构建工作，无需手动逐一执行命令。
调用 Pyserini 索引工具: 它使用了 pyserini.index.lucene 模块，这是 Pyserini 提供的基于 Apache Lucene 的索引构建工具。
配置索引参数: 脚本为索引过程配置了一系列重要的参数，这些参数决定了索引的类型和包含的信息。
代码解析 (run_indexing.sh)
这个脚本由四个相似的命令块组成，我们以第一个为例进行详细解析，其他块的逻辑完全相同，只是输入和输出目录不同。
#!/bin/bash
# --- 为 100 个产品的数���集构建索引 ---
# 调用 Pyserini 的 Lucene 索引模块。
python -m pyserini.index.lucene \
# --collection: 指定要索引的文档集合类型。
# JsonCollection 表示输入数据是一系列 JSON 对象（通常是 .jsonl 文件）。
--collection JsonCollection \
# --input: 指定包含输入文件的目录。
# Pyserini 会自动在该目录下查找名为 documents.jsonl 的文件。
--input resources_100 \
# --index: 指定生成的索引文件将要存放的目录。
--index indexes_100 \
# --generator: 指定文档生成器，它负责解析输入文件并生成 Lucene 文档。
# DefaultLuceneDocumentGenerator 是处理标准 JsonCollection 格式的默认生成器。
--generator DefaultLuceneDocumentGenerator \
# --threads: 指定用于索引的线程数。这里使用 1 个线程。
--threads 1 \
# --storePositions: 指示索引器存储词元（token）的位置信息。
# 这对于短语查询和邻近查询非常重要。
--storePositions \
# --storeDocvectors: 指示索引器存储文档向量。
# 这对于一些高级的相似度计算（如 "More Like This"）很有用。
--storeDocvectors \
# --storeRaw: 指示索引器存储原始的文档内容（即完整的 JSON 字符串）。
# 这使得在搜索后可以从索引中直接恢复整个文档，而无需再次访问原始文件。
--storeRaw
# --- (以下是为 1k, 10k, 50k 数据集构建索引的重复命令块) ---
python -m pyserini.index.lucene \
--collection JsonCollection \
--input resources_1k \
--index indexes_1k \
# ... (其他参数同上)
python -m pyserini.index.lucene \
--collection JsonCollection \
--input resources_10k \
--index indexes_10k \
# ... (其他参数同上)
python -m pyserini.index.lucene \
--collection JsonCollection \
--input resources_50k \
--index indexes_50k \
# ... (其他参数同上)总结: convert_product_file_format.py 和 run_indexing.sh 共同构成了项目的数据准备和索引构建阶段。前者负责将原始数据清洗、转换并格式化为适合搜索引擎的 documents.jsonl 文件；后者则利用这些文件，通过 Pyserini 构建出高效的、多层次的搜索索引。这两个步骤是实现 Agent 强大的 search 工具功能不可或缺的前提。
normalize.py 功能详解
shared_libraries/web_agent_site/normalize.py 是一个辅助模块，主要负责数据规范化，特别是针对那些变化多端、格式不一的文本属性，如颜色和尺寸。数据规范化的目的是将非结构化的、自由格式的文本映射到一个预定义的、有限的、标准化的集合中。
这在数据分析和机器学习中非常重要，因为：
减少特征维度: 将上百种不同的颜色描述（如 "light blue", "sky blue", "baby blue"）都归一化为 "blue"，可以大大减少特征的维度，使模型更容易学习。
提高匹配准确性: 在计算奖励或匹配用户需求时，如果用户想要 "blue"，而产品是 "sky blue"，直接字符串比较会失败。但如果都归一化为 "blue"，就可以成功匹配。
该文件的主要功能是：
颜色规范化 (normalize_color): 提供一个函数，接收一个颜色字符串，并尝试将其映射到一个预定义的 COLOR_SET 中的标准颜色。
尺寸规范化 (normalize_color_size 中的尺寸部分): 提供一个函数，遍历所有产品尺寸，并使用一组正则表达式 (SIZE_PATTERNS) 和预定义尺寸集合 (SIZE_SET) 将它们归类。例如，"5ft", "5 feet" 都会被归类到与 ft 或 feet 相关的模式。
代码解析 (normalize.py)
# 导入 re 库用于正则表达式操作。
import re
from typing import Tuple
# 定义一个包含标准颜色名称的列表。
# 这是一个预定义的、有限的颜色集合，用于作为规范化的目标。
COLOR_SET = [
"alabaster", "apricot", "aqua", "ash", "asphalt", "azure", "banana", "beige",
"black", "blue", "blush", "bordeaux", "bronze", "brown", "burgundy", "camel",
# ... (省略了大部分颜色)
"white", "wine", "yellow",
]
# 定义一个包含标准尺寸名称和模式的列表。
SIZE_SET = [
"xx-large", "3x-large", "4x-large", "5x-large", "x-large", "x-small",
"medium", "large", "small", "queen", "twin", "full", "king", "one size", "pack",
]
# 定义一组正则表达式，用于匹配各种非标准的尺寸描述。
SIZE_PATTERNS = [
re.compile(r"(.*)neck(.*)sleeve"), # 如 "15 neck 32 sleeve"
re.compile(r"(.*)w x(.*)l"), # 如 "32w x 30l"
re.compile(r"(.*)inch"), # 如 "5 inch"
re.compile(r"(\d+)\"$"), # 如 "5"" (5英寸)
# ... (省略了大部分模式)
]
# 将 SIZE_SET 中的每个字符串也编译成正则表达式，并与 SIZE_PATTERNS 合并。
SIZE_PATTERNS = [re.compile(s) for s in SIZE_SET] + SIZE_PATTERNS
# 函数：颜色规范化
def normalize_color(color_string: str) -> str:
"""
接收一个颜色字符串，并尝试提取其中包含的第一个标准颜色。
例如，输入 "light blue" 会返回 "blue"。
"""
# 遍历标准颜色集合。
for norm_color in COLOR_SET:
# 如果标准颜色是输入字符串的子串。
if norm_color in color_string:
# 返回找到的第一个标准颜色。
return norm_color
# 如果没有在输入字符串中找到任何标准颜色，则返回原始字符串。
return color_string
# 函数：颜色和尺寸的批量规范化 (这个函数似乎在项目中未被直接调用，但展示了方法)
def normalize_color_size(product_prices: dict) -> Tuple[dict, dict]:
"""
遍历所有产品，为所有的颜色和尺寸创建到其规范化值的映射。
返回两个字典：color_mapping 和 size_mapping。
"""
# ... (省略了具体的实现代码)
# 这个函数的逻辑是：
# 1. 从所有产品中提取出所有独特的颜色和尺寸字符串。
# 2. 对每个独特的颜色字符串，使用类似于 normalize_color 的逻辑找到其标准映射。
# 3. 对每个独特的尺寸字符串，遍历 SIZE_PATTERNS，找到第一个匹配的模式作为其标准映射。
# 4. 返回 color_mapping 和 size_mapping 两个字典。
# ...
return color_mapping, size_mappinggoal.py 功能详解
shared_libraries/web_agent_site/goal.py 文件是定义 Agent 的任务目标 和 评估其任务完成度（即奖励） 的核心业务逻辑模块。它回答了两个关键问题：“Agent 应该做什么？” 和 “Agent 做得好不好？”。
主要功能包括：
目标生成 (get_goals, get_human_goals, get_synthetic_goals):
目标 (Goal) 是一个数据结构，它精确定义了一次购物任务。通常包含：
asin: 目标商品的唯一 ID。
instruction_text: 给 Agent 的自然语言指令，如“我想要一件红色的纯棉 T 恤，价格低于 20 美元”。
attributes: 期望的商品属性列表，如 ['cotton', 'soft']。
goal_options: 期望的购买选项，如 {'color': 'red', 'size': 'medium'}。
price_upper: 价格上限。
get_human_goals: 从人类标注的数据文件 (items_human_ins.json) 中加载目标。这些目标通常更自然、更符合真实用户的表达方式。
get_synthetic_goals: 从产品数据中自动合成目标。它会遍历产品的各种属性和选项组合，生成大量的、结构化的任务目标。这对于大规模的自动化测试和训练非常有用。
奖励计算 (get_reward):
这是评估 Agent 行为的核心函数。当 Agent 完成一次购买（点击 "Buy Now"）后，该函数会被调用。
它将 Agent 实际购买的商品 与预设的 目标 (Goal) 进行详细比较，并计算出一个从 0.0 到 1.0 的分数。
多维度、细粒度的奖励机制:
r_type (类型奖励): 比较购买商品和目标商品在类别、查询词、标题上的相似度。如果 Agent 买了一双鞋，但目标是一件衣服，这个分数会很低。
r_price (价格奖励): 判断购买价格是否低于目标价格上限。
r_att (属性奖励): 比较购买商品的属性（从标题、描述、特性中提取）与目标属性的匹配度。使用了 thefuzz 库进行模糊字符串匹配，增加了鲁棒性。
r_option (选项奖励): 比较购买时选择的颜色、尺寸等选项与目标选项的匹配度。
最终的总奖励是这些分项奖励的加权组合，这种设计使得奖励信号非常丰富，能够精确地反映 Agent 在任务的哪个方面做得好或不好。
代码解析 (goal.py)
# 导入所需库
from collections import defaultdict
import itertools
import random
from rich import print # 用于美化打印输出
import spacy # 用于自然语言处理，如词性标注
from thefuzz import fuzz # 用于模糊字符串匹配
from .normalize import normalize_color # 导入颜色规范化函数
# 加载 spacy 的英文模型
nlp = spacy.load("en_core_web_sm")
# 定义预设的价格范围，用于生成合成目标中的价格上限
PRICE_RANGE = [10.0 * i for i in range(1, 100)]
# 函数：获取目标 (主入口)
def get_goals(all_products, product_prices, human_goals=True):
# 根据 human_goals 标志，决定是加载人类标注的目标还是自动合成的目标。
if human_goals:
return get_human_goals(all_products, product_prices)
else:
return get_synthetic_goals(all_products, product_prices)
# 函数：加载人类标注的目标
def get_human_goals(all_products, product_prices):
# ... (实现细节省略)
# 核心逻辑：
# 1. 遍历所有产品。
# 2. 如果产品在人类标注文件中有对应的 "instructions"，则为每条指令创建一个 goal 对象。
# 3. goal 对象包含 asin, category, query, name, product_category, instruction_text,
# attributes, price_upper, goal_options 等字段。
# 4. 价格上限是基于产品实际价格随机生成的。
# 5. 返回所有生成的目标列表。
# ...
return goals
# 函数：自动合成目标
def get_synthetic_goals(all_products, product_prices):
# ... (实现细节省略)
# 核心逻辑：
# 1. 遍历所有产品。
# 2. 对每个产品，获取其所有可能的选项组合（如 color 和 size 的所有笛卡尔积）。
# 3. 对每一种选项组合，结合产品的基本属性和指令文本，生成一个 goal 对象。
# 4. 这种方法可以为同一个产品生成多个不同的、精细化的购物目标。
# ...
return goals
# 函数：计算类型奖励
def get_type_reward(purchased_product, goal):
"""判断购买的商品是否与目标商品属于同一大类"""
# 比较 query、product_category 和 name (标题) 的相似度。
# 使用 spacy 提取标题中的名词进行比较，以提高准确性。
# ...
# 返回一个字典，包含 r_type (0.0-1.0 的分数) 和各项匹配的布尔值。
return dict(r_type=r_type, query_match=query_match, category_match=category_match, title_score=title_score)
# 函数：计算属性奖励
def get_attribute_reward(purchased_product, goal):
"""计算购买商品的属性与目标属性的匹配程度"""
purchased_attrs = purchased_product["Attributes"]
goal_attrs = goal["attributes"]
num_attr_matches = 0
# 遍历每个目标属性。
for g_attr in goal_attrs:
matched = False
# 使用模糊匹配 (fuzz.token_set_ratio) 在购买商品的属性列表中查找。
for p_attr in purchased_attrs:
score = fuzz.token_set_ratio(p_attr, g_attr)
if score > 85: # 设置一个较高的相似度阈值
num_attr_matches += 1
matched = True
break
# 如果在属性列表中未找到，则在标题、描述、要点中再次查找。
if not matched and (g_attr in purchased_product["Title"].lower() or ...):
num_attr_matches += 1
# 奖励 = 匹配上的目标属性数量 / 总目标属性数量
r_attr = num_attr_matches / len(goal_attrs) if goal_attrs else 1.0
return r_attr, num_attr_matches
# 函数：计算选项奖励
def get_option_reward(purchased_options, goal_options):
"""计算购买时选择的选项与目标选项的匹配程度"""
# 对选项值进行规范化（例如，颜色规范化）。
purchased_options = [normalize_color(o) for o in purchased_options]
goal_options = [normalize_color(o) for o in goal_options]
# 使用模糊匹配计算匹配上的选项数量。
# ...
# 奖励 = 匹配上的目标选项数量 / 总目标选项数量
r_option = num_option_matches / len(goal_options) if goal_options else 1.0
return r_option, num_option_matches
# 函数：计算总奖励 (主入口)
def get_reward(purchased_product, goal, price, options, **kwargs):
"""计算购买行为的最终综合得分"""
# 1. 调用 get_type_reward 计算类型奖励。
r_type_dict = get_type_reward(purchased_product, goal)
# 2. 计算价格奖励。
r_price = (price <= goal["price_upper"]) if goal["price_upper"] > 0 else 1.0
# 3. 调用 get_attribute_reward 计算属性奖励。
r_att, num_attr_matches = get_attribute_reward(purchased_product, goal)
# 4. 调用 get_option_reward 计算选项奖励。
r_option, num_option_matches = get_option_reward(list(options.values()), list(goal["goal_options"].values()))
# 5. 将各分项奖励加权求和。
# 这里的权重是基于各项目标的数量（属性数量、选项数量、价格（计为1项））。
total_reward = (num_attr_matches + num_option_matches + int(r_price)) / (
len(goal["attributes"]) + len(goal["goal_options"]) + 1
)
# 6. 最终奖励要乘以类型奖励。如果买错了大类，即使属性选项都对，得分也会很低。
total_reward *= r_type_dict["r_type"]
# 如果需要，返回详细的奖励分项信息。
if kwargs.get("verbose", False):
info = { ... }
return total_reward, info
return total_rewardengine.py 功能详解
engine.py 是 WebShop 模拟环境的核心引擎。它包含了驱动环境运行的各种核心函数和数据结构，是连接数据、搜索引擎、HTML 模板和服务器逻辑的枢杻。
主要功能包括：
动作到页面的映射 (map_action_to_html):
这是动态页面渲染的核心。它接收一个动作（如 "search", "click[Buy Now]"）和一系列上下文参数（如产品信息、搜索结果）。
根据动作类型，它会选择正确的 HTML 模板文件（如 results_page.html, done_page.html）。
使用 Flask 的 render_template_string 函数，将上下文数据填充到模板中，生成最终的 HTML 页面字符串。
数据加载与预处理 (load_products):
负责从原始的 items_shuffle.json 文件中加载产品数据。
在加载过程中进行了一系列的数据清洗和预处理：
清理无用的键。
处理价格字符串，将其转换为数值。
解析复杂的 customization_options 字段，将其转换为更简洁的 options 字典。
关联和处理属性、评论等信息。
返回一个干净、随时可用的产品列表、产品字典（ASIN -> product）和价格字典。
搜索引擎集成 (init_search_engine, get_top_n_product_from_keywords):
init_search_engine: 根据配置的产品数量，加载相应的 Pyserini/Lucene 搜索索引，返回一个 LuceneSearcher 对象。
get_top_n_product_from_keywords: 接收关键词列表和 LuceneSearcher 对象，执行搜索，并返回匹配度最高的 N 个产品列表。
辅助工具函数:
parse_action: 解析动作字符串，如将 "search[tv shows]" 解析为 ("search", "tv shows")。
get_product_per_page: 实现搜索结果的分页逻辑。
read_html_template: 读取 HTML 模板文件的内容。
generate_product_prices: 为每个产品生成一个随机但固定的价格。
代码解析 (engine.py)
# 导入所需库
from ast import literal_eval
from collections import defaultdict
from decimal import Decimal
import json
import os
import random
import re
from flask import render_template_string # 用于渲染 HTML 模板
from pyserini.search.lucene import LuceneSearcher # Pyserini 搜索引擎
from rich import print
from tqdm import tqdm
from ..utils import ( # 导入路径等常量
BASE_DIR,
DEFAULT_ATTR_PATH,
HUMAN_ATTR_PATH,
)
# 定义模板目录和一些常量
TEMPLATE_DIR = os.path.join(BASE_DIR, "templates")
# ... (END_BUTTON, NEXT_PAGE 等常量)
# 动作名称到其对应 HTML 模板文件的映射
ACTION_TO_TEMPLATE = {
"Description": "description_page.html",
"Features": "features_page.html",
"Reviews": "review_page.html",
"Attributes": "attributes_page.html",
}
# 函数：动作到页面的映射
def map_action_to_html(action, **kwargs):
"""根据动作和上下文参数，选择模板并渲染成 HTML 页面"""
action_name, action_arg = parse_action(action)
# 根据 action_name 和 action_arg 选择不同的分支
if action_name == "start":
path = os.path.join(TEMPLATE_DIR, "search_page.html")
# 使用 render_template_string 填充模板
html = render_template_string(read_html_template(path), **kwargs)
elif action_name == "search":
path = os.path.join(TEMPLATE_DIR, "results_page.html")
html = render_template_string(read_html_template(path), **kwargs)
elif action_name == "click" and action_arg == END_BUTTON:
path = os.path.join(TEMPLATE_DIR, "done_page.html")
html = render_template_string(read_html_template(path), **kwargs)
elif action_name == "click" and action_arg in ACTION_TO_TEMPLATE:
path = os.path.join(TEMPLATE_DIR, ACTION_TO_TEMPLATE[action_arg])
html = render_template_string(read_html_template(path), **kwargs)
elif action_name == "click":
path = os.path.join(TEMPLATE_DIR, "item_page.html")
html = render_template_string(read_html_template(path), **kwargs)
else:
raise ValueError("Action name not recognized.")
return html
# 函数：解析动作字符串
def parse_action(action):
"""将 "name[arg]" 格式的字符串解析为 (name, arg)"""
# ... (使用正则表达式进行解析)
return action_name, action_arg
# 函数：初始化搜索引擎
def init_search_engine(num_products=None):
"""根据产品数量加载相应的 Pyserini/Lucene 搜索索引"""
# 根据 num_products 选择不同的索引目录
if num_products == 100:
indexes = "indexes_100"
# ... (其他规模)
else:
indexes = "indexes_1k" # 默认值
# 创建并返回一个 LuceneSearcher 实例
search_engine = LuceneSearcher(os.path.join(BASE_DIR, f"../search_engine/{indexes}"))
return search_engine
# 函数：执行搜索
def get_top_n_product_from_keywords(keywords, search_engine, ...):
"""使用搜索引擎执行关键词搜索，并返回 top N 结果"""
# ... (处理特殊关键词如 "<r>" 随机, "<a>" 属性, "<c>" 类别)
# else: # 普通关键词搜索
keywords_str = " ".join(keywords)
# 调用 search_engine.search() 执行搜索
hits = search_engine.search(keywords_str, k=SEARCH_RETURN_N)
# 从搜索结果中提取文档 ID (即 ASIN)
top_n_asins = [json.loads(search_engine.doc(hit.docid).raw())["id"] for hit in hits]
# 根据 ASIN 列表从产品字典中获取完整的产品信息
top_n_products = [product_item_dict[asin] for asin in top_n_asins if asin in product_item_dict]
return top_n_products
# 函数：加载和预处理产品数据
def load_products(filepath, num_products=None, human_goals=True):
"""从 JSON 文件加载产品数据，并进行清洗和预处理"""
with open(filepath) as f:
products = json.load(f)
print("Products loaded.")
# ... (调用 clean_product_keys 清理无用字段)
# 如果指定了 num_products，则只取前 N 个产品
if num_products is not None:
products = products[:num_products]
# 遍历每个产品，进行详细的字段处理和转换
for i, p in tqdm(enumerate(products), total=len(products)):
# ... (处理 Title, Description, Reviews, BulletPoints)
# ... (处理 pricing 字符串，转换为数值)
# ... (处理 customization_options，转换为简洁的 options 字典)
# ... (关联 attributes 和 instructions)
# ...
# 返回处理好的产品列表、产品字典和价格字典
return all_products, product_item_dict, product_prices, attribute_to_asinsutils.py 功能详解
shared_libraries/web_agent_site/utils.py 在项目中扮演着工具箱和常量库的角色。它通常包含那些在项目多个不同模块中都可能被用到的、与核心业务逻辑不直接相关的辅助函数和常量定义。将这些内容集中在 utils.py 中，有助于保持代码的整洁、避免重复，并方便管理。
该文件的主要功能可以分为以下三类：
定义项目范围内的常量:
文件路径: 文件中定义了大量指向关键数据文件的绝对路径常量。例如：
BASE_DIR: 获取当前文件所在的目录，作为计算其他路径的基准。
DEFAULT_FILE_PATH: 指向主要的产品数据文件 (items_shuffle.json)。
DEFAULT_ATTR_PATH, HUMAN_ATTR_PATH: 指向包含产品属性和人类标注指令的 JSON 文件。
FEAT_CONV, FEAT_IDS: 指向用于图像搜索的预计算特征文件。
这样做的好处: 将所有硬编码的路径集中管理。如果未来文件位置发生变化，只需修改 utils.py 这一个地方，而不需要在整个项目中去查找和替换，大大提高了代码的可维护性。
提供通用的辅助函数:
random_idx(cum_weights): 这是一个实现带权重随机抽样的函数。在项目中，不同的购物目标（Goal）可能有不同的权重（例如，一些更常见或更重要的目标权重更高）。这个函数可以根据这些权重，公平地随机选择一个目标。它的工作原理是：
在累积权重的总和范围内生成一个随机数。
使用 bisect.bisect 在累积权重列表中快速找到这个随机数应该插入的位置，该位置的索引就是被选中的目标的索引。这是一种非常高效的带权重抽样算法。
setup_logger(session_id, user_log_dir): 这是一个设置日志记录器的函数。它可以为每个会话（由 session_id 标识）创建一个独立的日志文件。这对于追踪和调试单个用户的完整交互流程非常有用，可以避免所有日志混杂在一起。
generate_mturk_code(session_id): 这个函数用于为 Amazon Mechanical Turk (MTurk) 任务生成一个唯一的完成码。MTurk 是一个众包平台，研究者经常用它来收集人类与 Agent 交互的数据。当一个“工人”（Turker）完成一次交互任务后，系统会生成一个基于 session_id 的、独一无二的验证码，工人可以凭此码获得报酬。这确保了任务完成的有效性和可追溯性。
调试开关:
DEBUG_PROD_SIZE = None: 这是一个调试开关。在开发过程中，可以将其设置为一个较小的数字（如 100），然后在代码的其他地方（如 load_products 函数中）检查这个值，如果它不是 None，就只加载指定数量的产品。这可以极大地加快程序的启动和调试速度。在最终发布时，将其设置回 None 即可加载完整数据集。
总结：utils.py 是一个典型的项目辅助模块。它通过集中管理常量和提供可重用的工具函数，简化了其他模块的代码，提高了整个项目的代码质量和可维护性。它就像一个多功能工具箱，为项目的各个部分提供了必要的“螺丝刀”、“扳手”和“卷尺”。
# 导入所需库
import bisect # 用于实现高效的二分查找，这里用于带权重随机抽样。
import hashlib # 用于生成哈希值，这里用于创建 MTurk 验证码。
import logging # 用于设置和管理日志记录。
from os.path import abspath, dirname, join # 用于处理文件和目录路径。
import random # 用于生成随机数。
# --- 1. 定义项目范围内的常量 ---
# 获取当前 utils.py 文件所在的目录的绝对路径。
# 这将作为计算项目中其他所有文件路径的基准。
BASE_DIR = dirname(abspath(__file__))
# 调试开关：用于在开发时限制加载的产品数量，以加快启动速度。
# 设置为 None 表示不限制（加载全部）；设置为数字（如 100）则只加载相应数量。
DEBUG_PROD_SIZE = None
# 定义指向各种数据文件的路径常量。
# 使用 join 函数可以确保路径在不同操作系统（Windows, Linux, macOS）下都是正确的。
DEFAULT_ATTR_PATH = join(BASE_DIR, "../data/items_ins_v2.json") # 默认属性和合成指令文件
DEFAULT_FILE_PATH = join(BASE_DIR, "../data/items_shuffle.json") # 主要的产品数据文件
DEFAULT_REVIEW_PATH = join(BASE_DIR, "../data/reviews.json") # 评论数据文件
FEAT_CONV = join(BASE_DIR, "../data/feat_conv.pt") # 预计算的图像特征文件
FEAT_IDS = join(BASE_DIR, "../data/feat_ids.pt") # 图像特征对应的 ID 文件
HUMAN_ATTR_PATH = join(BASE_DIR, "../data/items_human_ins.json") # 人类标注的指令文件
# (下面这行是重复的，可以删除)
# HUMAN_ATTR_PATH = join(BASE_DIR, "../data/items_human_ins.json")
# --- 2. 提供通用的辅助函数 ---
def random_idx(cum_weights):
"""
根据权重的累积分布进行带权重随机抽样。
Args:
cum_weights (list): 一个权重的累积和列表，例如 [0, w1, w1+w2, w1+w2+w3, ...]。
Returns:
int: 被选中的项的索引。
"""
# 在 0 到总权重（即累积权重列表的最后一个元素）之间生成一个随机浮点数。
pos = random.uniform(0, cum_weights[-1])
# 使用二分查找，快速找到这个随机数在累积权重列表中的位置。
# bisect.bisect 返回的索引即为被抽中的项的索引。
idx = bisect.bisect(cum_weights, pos)
# 确保返回的索引不会越界。
idx = min(idx, len(cum_weights) - 2)
return idx
def setup_logger(session_id, user_log_dir):
"""
为给定的会话 ID 创建并配置一个专用的日志记录器。
Args:
session_id (str): 当前交互的会话 ID。
user_log_dir (Path object): 用于存放日志文件的目录。
Returns:
logging.Logger: 一个配置好的日志记录器实例。
"""
# 获取一个以 session_id 命名的日志记录器实例。
logger = logging.getLogger(session_id)
# 定义日志消息的格式。
formatter = logging.Formatter("%(message)s")
# 创建一个文件处理器，将日志写入到 {session_id}.jsonl 文件中。
file_handler = logging.FileHandler(user_log_dir / f"{session_id}.jsonl", mode="w")
# 为文件处理器设置格式。
file_handler.setFormatter(formatter)
# 设置日志记录器的级别为 INFO，即只记录 INFO 及以上级别的日志。
logger.setLevel(logging.INFO)
# 将文件处理器添加到日志记录器中。
logger.addHandler(file_handler)
return logger
def generate_mturk_code(session_id: str) -> str:
"""
为给定的会话 ID 生成一个唯一的 Amazon Mechanical Turk (MTurk) 验证码。
Args:
session_id (str): 需要为其生成验证码的会话 ID。
Returns:
str: 一个 10 位的大写字母/数字组成的验证码。
"""
# 使用 SHA1 哈希算法处理会话 ID 字符串。
# .encode() 是必需的，因为哈希函数处理的是字节串。
sha = hashlib.sha1(session_id.encode())
# 获取哈希值的十六进制表示，并取前 10 个字符，然后转换为大写。
return sha.hexdigest()[:10].upper()HTML模板文件(search_page.html，review_page.html，results_page.html，item_page.html，features_page.html，done_page.html，description_page.html，attributes_page.html)功能详解
这些文件共同构成了 WebShop 模拟环境的用户界面 (UI) 模板。它们是动态的，使用了 Jinja2 模板引擎语法（例如 {{ variable }} 和 {% for item in items %}），在运行时由 engine.py 中的 map_action_to_html 函数填充真实数据，最终生成用户（或 Agent）“看到”的 HTML 页面。
1. search_page.html - 搜索首页
功能: 这是用户与 WebShop 交互的入口页面。
品牌/Logo 展示: 显示 "WebShop" 的标题，建立品牌认知。
指令显示:
通过 <h4>Instruction:<br>{{ instruction_text }}</h4> 显示当前的购物任务指令。这对于指导 Agent 或人类用户完成任务至关重要。
搜索输入框:
提供一个文本输入框 (<input id="search_input" ...>)，让用户可以输入想要搜索的商品关键词。
搜索按钮:
提供一个带有搜索图标的按钮 (<button class="btn btn-success" ...>Search</button>)，用户点击后会提交搜索表单。
表单提交: 整个搜索框和按钮被包裹在一个 <form> 标签中，它会将用户输入的 search_query POST 到 url_for('index', session_id=session_id)，这个请求会被 SimServer 捕获并分发到搜索逻辑。
总结: 这是整个购物流程的起点，提供最基本的任务展示和搜索功能。
2. results_page.html - 搜索结果页
功能: 在用户执行搜索后，展示匹配的商品列表。
页面信息: 显示当前页码和总结果数 (<h3>Page {{page}} (Total results: {{total}})</h3>)，让用户了解搜索的广度。
分页导航:
提供 "< Prev" 和 "Next >" 按钮，允许用户在多个搜索结果页面之间切换。
通过 {% if page > 1 %} 等逻辑判断，仅在适当的时候显示这些按钮（例如，第一页不显示 "Prev"）。
商品列表:
使用 {% for item in products %} 循环遍历后端传入的 products 列表。
商品信息展示: 对每个商品，它会展示：
图片: <img src="{{item.MainImage}}" ...>
ASIN (作为链接): <a class="product-link" ...>{{item.asin}}</a>。这个链接的 href 指向该商品的详情页 (item_page)，是导航到下一步的关键。
标题: <h4>{{item.Title}}</h4>
价格: <h5>{{item.Price}}</h5>
返回搜索: 提供一个 "Back to Search" 按钮，允许用户随时放弃当前搜索，返回到 search_page.html。
总结: 这是连接搜索和具体商品的桥梁，提供了核心的商品浏览和分页功能。
3. item_page.html - 商品详情主页
功能: 展示单个商品的详细信息，并提供所有交互选项，是用户决策的核心页面。
导航按钮:
"Back to Search": 返回搜索首页。
"< Prev": 返回到之前的搜索结果页 (results_page.html)。
核心商品信息:
主图: <img id="product-image" src="{{product_info.MainImage}}" ...>。这张图片可以根据用户选择的选项动态变化（通过 JavaScript）。
标题: <h2>{{product_info.Title}}</h2>
价格: <h4>Price: {{product_info.Price}}</h4>
评分: <h4>Rating: {{product_info.Rating}}</h4>
购买选项 (Options):
这是此页面最复杂的部分。使用 {% for option_name, option_contents in product_info.options.items() %} 遍历所有选项类型（如 "color", "size"）。
对每种选项类型，再用一个循环创建一组单选按钮 (radio buttons)，如 <input type="radio" ...>。
动态 URL: 每个选项按钮的 data-url 属性包含一个特殊的 URL。这个 URL 是在选择该选项后重新加载当前页面的 URL，并且 URL 中包含了新选择的选项。
JavaScript 交互:
页面底部的 <script> 负责处理选项交互。
当页面加载时，它会根据传入的 options 变量，自动勾选上已经选择的选项。
它会监听单选按钮的点击事件，当用户点击一个新选项时，window.location.href = this.dataset.url; 会触发页面跳转到对应的 data-url，从而实现无刷新感的页面状态更新。
它还能根据选择的选项动态更新产品主图。
子信息页导航:
提供 "Description", "Features", "Reviews", 和 (可选的) "Attributes" 按钮。
每个按钮都是一个独立的表单，点击后会将用户导航到对应的子信息页面（如 description_page.html）。
购买按钮:
提供一个醒目的 "Buy Now" 按钮，点击后将导航到 done_page.html，完成购买流程。
总结: 这是信息最密集、交互最复杂的页面，集成了商品展示、选项选择、信息导航和最终购买决策功能。
4. description_page.html, features_page.html, attributes_page.html, review_page.html - 子信息页
description_page.html：
features_page.html：
attributes_page.html：
review_page.html：
功能: 这些页面结构非常相似，它们分别用于展示商品的某一个特定方面的信息，以保持主详情页的简洁性。
共同结构:
导航: 都包含 "Back to Search" 和 "< Prev" 按钮。关键是 "< Prev" 按钮，它允许用户从子信息页返回到 item_page.html。
指令显示: 同样显示任务指令。
内容展示:
description_page.html: 显示 <p class="product-info">{{product_info.Description}}</p>。
features_page.html: 使用 <ul> 和 <li> 列表展示 product_info.BulletPoints。
attributes_page.html: 使用 <ul> 和 <li> 列表展示 product_info.Attributes。
review_page.html: 结构最复杂，循环遍历 product_info.Reviews，对每条评论展示标题、星级评分（通过循环 fa-star 图标实现）和评论正文。
总结: 这些是 item_page.html 的附属页面，提供了更深层次的信息，Agent 需要通过在这些页面和主详情页之间来回导航来收集完整信息。
5. done_page.html - 完成/结算页
功能: 这是购物流程的终点，用于告知用户购买结果并展示得分。
感谢信息: 显示 "Thank you for shopping with us!"。
MTurk 验证码: 显示 <pre>{{ mturk_code }}</pre>，供众包平台的测试者提交以获取报酬。
得分显示: 最重要的部分是 <h3>Your score ...<pre>{{ reward }}</pre></h3>，它向用户或 Agent 展示了这次购物任务的最终得分。
调试信息 (隐藏):
页面中包含一个 <div style="display:none"> 块。
这个块里面包含了极其详细的调试信息，例如：
购买的商品 (purchased_attrs, purchased-category 等)。
目标商品的信息 (goal-asin, goal-options 等)。
完整的 goal 对象和 reward_info 字典。
这些信息在页面上是不可见的，但存在于 HTML 源码中。这对于开发者分析 Agent 的行为（为什么得分高/低）提供了宝贵的数据。
总结: 这是流程的结束和反馈页面，提供了任务完成的确认、得分反馈以及丰富的后台调试信息。
总结
个性化购物 Agent 是一个高度集成且功能完备的端到端智能代理系统。它不仅仅是一个聊天机器人，而是一个能够在模拟的电子商务环境 (WebShop) 中自主执行复杂、多步骤任务的智能体 (Agent)。项目的核心目标是模拟一个专业的购物助手，从理解用户模糊的购物需求开始，到主动搜索、比较商品、引导决策，并最终完成模拟购买的全过程。
该项目巧妙地将当前 AI Agent 领域的几大核心技术融合在一起，构成了一个清晰、可扩展的架构。
核心架构与技术支柱
整个项目可以被看作是由四大支柱构成的：
1. 智能核心 (The Brain): 大型语言模型 (LLM) 与提示工程
大脑: 项目的核心决策能力由一个大型语言模型（如 Grok-3 Beta）提供，通过 agent.py 中的 LiteLlm 包装器集成。
灵魂: prompt.py 中定义的详细系统提示是 Agent 的“灵魂”。它为 LLM 设定了明确的角色、目标、交互流程和行为约束，将一个通用的模型“塑造”成一个专业的购物专家。
2. 模拟世界 (The World): WebShop 环境
操场: 项目最复杂也最亮眼的部分是其自包含的模拟 WebShop 环境。web_agent_text_env.py 定义了一个遵循 OpenAI Gym 规范的环境，其中 SimServer 模拟后端逻辑，SimBrowser 模拟前端行为。
动态渲染: engine.py 作为引擎，驱动着整个环境的运转。它负责加载数据、集成搜索引擎，并根据 Agent 的动作，使用 templates/ 目录下的 HTML 模板（如 search_page.html, item_page.html 等）动态渲染出 Agent “看到”的页面。
3. 交互工具 (The Hands and Eyes): Search 与 Click
手和眼: Agent 通过 search.py 和 click.py 中定义的工具与模拟世界进行交互。search 工具让 Agent 能够“看”到环境中的商品，而 click 工具则让它能够“触摸”和“操作”页面元素。
桥梁: 这些工具是连接 LLM 的抽象决策（“我需要搜索裙子”）和环境具体实现（webshop_env.step("search[裙子]")）的关键桥梁。
4. 数据与规则 (The Foundation and Rulebook)
知识库: 项目的基础是庞大的产品数据集 (items_*.json)。convert_product_file_format.py 和 run_indexing.sh 协同工作，将这些原始数据处理成 Pyserini/Lucene 搜索引擎可以高效检索的索引，为 search 工具提供了动力。
评价体系: goal.py 和 normalize.py 定义了项目的“规则手册”。goal.py 负责生成明确的购物目标，并提供一个极其精细的奖励函数 (get_reward)。这个函数能够从类型、属性、选项、价格等多个维度综合评估 Agent 任务的完成度，为 Agent 的性能评估和潜在的强化学习提供了高质量的反馈信号。
完整工作流程
一个典型的交互流程如下：
初始化: init_env.py 创建并共享一个全局的、包含 50,000 个商品的 webshop_env 实例。
用户请求: 用户向 Agent 提出购物需求。
Agent 决策: LLM (在 prompt.py 指导下) 决定调用 search 工具。
搜索与观察: search 工具在 webshop_env 中执行搜索，环境返回一个包含商品列表的 results_page.html 的文本表示。
浏览与导航: Agent 分析结果，与用户沟通，并根据选择调用 click 工具点击某个商品链接。
主动探索: 进入 item_page.html 后，Agent 根据指令，主动调用 click 工具在 description, features, reviews 等子页面之间导航，收集完整信息。
决策与购买: Agent 总结信息，与用户确认购买选项（如颜色、尺寸），并最终调用 click 点击 "Buy Now"。
评估与结束: 环境进入 done_page.html，goal.py 中的 get_reward 函数被调用，计算最终得分，流程结束。
总结
个性化购物 Agent 项目是一个杰出的 AI Agent 范例。它不仅展示了如何利用 LLM 进行复杂的任务规划，更重要的是，它提供了一套完整的、可工作的环境模拟、工具集成、数据处理和性能评估的蓝图。
其模块化的设计（Agent/环境/工具的分离）、高度仿真的交互流程、数据驱动的搜索引擎以及精细化的奖励机制，使其不仅是一个引人注目的演示，更是一个极具价值的学习资源和可用于进一步研究与开发的坚实基础。它清晰地揭示了构建能够解决现实世界复杂问题的下一代智能 Agent 所需的核心要素。