记录一下从 0 开始编写第一个 ACT
1 确定需求
1.1 想做什么
首先，根据半个宇宙的社区属性与方向，确定了开发的 ACT 最好需要贴近日常，于是根据我的一些生活习惯，决定实现「收集分析百度上的种草数据」功能。粗略浏览了一下社区，发现目前没有完全和我想法类似的 ACT，那么就以这个功能为核心开始着手开发。
在浏览社区的过程中，发现在 ACT 的具体开发上并没有很详细的说明或记录，那么就以此文来抛砖引玉吧，为社区发展尽一份力，也欢迎大家提供反馈和建议。
1.2 做到什么
考虑到单纯「种草」的范围太大了，于是给定一个 1.0 版本的支持搜索范围，并以此作为测试用例：
- 实现搜索：「深圳的美食」，并返回种草排行榜。
- 实现搜索：「打工人好物」，并返回种草排行榜。
2 前置准备
2.1 开发环境
文档中提到，ACT 基本可以使用 Python 语言开发，同时 kOS 的控制台具有代码运行与测试的功能，因此仅使用 VS Code 来提高本地代码的可读性，不考虑代码的编译环境。
2.2 AI 工具
由于我 没有 Python 语言的开发经验（完全没用过） ，所以为了提高效率，使用了一些 AI 工具。包括但不限于：
- 向 AI 提问一些简单的基础语法。
- 在群里要了一份开发手册的 PDF 版本，丢进 Kimi 大模型中解析然后提问。
- 在 VS Code 中配置「通义灵码」插件。
多管齐下协助开发。
3 功能实现
3.1 需求拆解
在我看来，对于这样的大模型应用层开发，完全可以仅把大模型当作分析文本的黑盒：只需输入需要分析的文本数据，并给定一些参数，他就能返回处理的结果。
因此，我们只需要将需求的各个功能抽象成不同模块，然后结合标准 ACT 的功能实现，将模块拆分为实现的步骤，即可完成需求拆解（其实这里也可以丢给 AI 来分解，但是当时忘了）。
这里我们直接一步到位，将需求拆分成这几个步骤：
- 获取用户输入内容
- 解析用户输入内容并拆分为元数据
- 将元数据分别进行网络搜索
- 整理搜索结果
- 语义分析
- 返回结果给用户
- 记录日志
3.2 具体实现
3.2.1 获取用户输入内容
直接参考官方给出的 ACT 模板：
# 获取ACT触发启动时的消息
query = nact.k_get_act_query()
if not query:
query = nact.k_ask_for_answer('请输入想种草的内容') # 如果没有查询要求，请求用户输入
3.2.2 解析用户输入内容并拆分为元数据
这一步是需求中的一个难点，也是此 ACT 与直接使用「搜一搜」不同的地方。这里我参考了开发者「可道玩AI」的「星伴 ACT：代表作」里的代表作搜索结果优化思路，对用户的提问内容做进一步加工。
这里暂时直接使用 nact.k_semantic_rephrase 来对语义进行转换，后续会更新更好的方案。
考虑了两个方案如下：
- 递归调用上一次重写的结果，也就是将上一次重写的结果作为上下文填入函数的参数中。
- 分别以「缩写」和「扩写」为要求重写两次，最终分别查询三次结果。
方案一：
search_list = [query]
count = 0
while count < 3:
if count != 0:
# 如果不是第一次循环，则在函数中加入上下文
search = nact.k_semantic_rephrase('在不影响本意的情况下重写这个问题', query, f'参考：{search}', '')
else:
# 第一次循环则直接调用函数
search = nact.k_semantic_rephrase('在不影响本意的情况下重写这个问题', query, '', '')
search_list.append(search)
count += 1
return search_list
测试结果如下：
方案二：
search_list = [query]
search = nact.k_semantic_rephrase('在不影响本意的情况下缩写这个问题', query, '', '')
search_list.append(search)
search = nact.k_semantic_rephrase('在不影响本意的情况下扩写这个问题', query, '', '')
search_list.append(search)
return search_list
测试结果如下：
使用搜索引擎分别测试搜索结果后，最终确定使用方案二作为该版本的实现方案，感兴趣的可以分别去搜索一下，这里就不放图了。
在多次测试方案后，进一步更新语义重写的「改写要求」参数。同时，在参考官方的「搜一搜」ACT 模板时，发现代码在处理搜索结果时使用了 KOSConcurrentBatcher 函数，因此更新方案代码，加入批处理，优化了该模块的运行时间。
更新代码如下：
# 创建批处理器
batcher: nact.KOSConcurrentBatcher = nact.k_concurrent_create_batcher()
# 方案二：
batcher.submit(nact.k_semantic_rephrase, '在不影响本意的情况下缩写这个问题', query, '', '')
batcher.submit(nact.k_semantic_rephrase, '在不影响本意的情况下，以提问的方式扩写这个问题', query, '', '')
# 等待批处理任务完成
results: list = batcher.wait()
for result in results:
if not result:
continue
# 获取批处理任务的结果
search_list.append(result)
return search_list
3.2.3 将元数据分别进行网络搜索
这里也是需求的难点之一，好在官方提供的「搜一搜」ACT 中已经有了完整的实现，照抄简单改一改即可。
部分代码如下：
# 创建批处理器
batcher: nact.KOSConcurrentBatcher = nact.k_concurrent_create_batcher()
tl_all = []
# 遍历问题，执行网络搜索
search_count = 0
for search in search_list:
tl = nact.k_web_search(search)
if not tl: # 没有搜索结果则跳过
search_count += 1
continue
for idx, topic in enumerate(tl): # 有结果则遍历搜索结果
tl_all.append(tl) # 将搜索结果添加到列表
batcher.submit(handle_topic, query, topic, idx) # 提交处理任务到批处理器
# 如果所有搜索都未搜索到结果则结束
if search_count == len(search_list):
nact.k_message_send('未搜索到相关内容！')
return None
# 等待批处理任务完成
results: list = batcher.wait()
handle_topic 函数如下：
# 定义处理单个搜索结果主题的函数
def handle_topic(query: str, topic: nact.KOSWebSearchTopic, idx: int) -> Tuple[nact.KOSWebPageData, str, str]:
# 获取网页数据
page_data: nact.KOSWebPageData = nact.k_web_get_page_data(topic.source_url)
if not page_data or not page_data.content:
return None, '', '' # 如果没有获取到数据或内容，返回空值
# 设置标题，如果主题没有标题，则使用网页数据的标题
title = topic.title if topic.title else page_data.title
if not title: # 如果还是没有标题
title = str(nact.k_time_cur_milliseconds()) + str(idx) # 使用当前时间毫秒数和索引作为标题
ref_tag = page_data.domain # 使用页面域名作为引用标签
else:
title = title.replace('/', '') # 替换掉标题中的斜杠
ref_tag = title # 标题作为引用标签
page_data.title = title # 更新网页数据的标题
# 对页面内容进行语义总结，生成摘要
summary = nact.k_semantic_summarize_answer(page_data.content, query)
return page_data, ref_tag, summary # 返回网页数据、引用标签和摘要
3.2.4 整理搜索结果
照抄后简单修改：
# 初始化引用源对象和摘要字符串
ref_source = nact.KOSMsgWebContentRefSource()
all_summary = ''
# 遍历批处理结果
for result in results:
if not result:
continue
(page_data, ref_tag, summary) = result
if not page_data:
continue
# 添加引用网页到引用源对象
ref_source.add_web_ref(ref_tag, page_data.url)
all_summary += summary + "\n" # 将摘要追加到字符串
3.2.5 语义分析
照抄：
# 对所有摘要进行语义总结
summary = nact.k_semantic_summarize_answer(all_summary, query)
if not summary:
nact.k_message_send('未搜索总结出有效内容') # 如果没有总结内容，发送消息告知用户
return
3.2.6 返回结果给用户
发送结果：
# 发送总结后的内容给用户
nact.k_message_send(summary, ref_source)
3.2.7 记录日志
遍历搜索结果，打印日志信息，保存记录：
# 打印日志信息
nact.k_print('开始记录种草结果')
# 再次遍历批处理结果，提交记录任务到批处理器
for idx, result in enumerate(results):
if not result:
continue
(page_data, ref_tag, summary) = result
if not page_data:
continue
batcher.submit(record_topic, all_tl[idx], page_data) # 提交记录主题的任务
# 等待记录任务完成
batcher.wait()
record_topic 函数如下：
# 定义记录搜索结果主题的函数
def record_topic(topic: nact.KOSWebSearchTopic, page_data: nact.KOSWebPageData):
# 获取当前本地日期字符串
time_str = nact.k_time_cur_local_day_str()
if nact.k_file_is_available(): # 检查文件系统是否可用
# 打开文件，准备将内容写入，如果文件存在则覆盖
file = nact.k_except_wrapper(nact.k_file_open, f'/种草/{time_str}_{page_data.title}.txt', nact.FileOpenMode.OVERWRITE,
source=nact.FileSource.WEB_LINK, source_url=topic.source_url)
if file: # 如果成功打开文件
nact.k_file_append(file, page_data.content) # 将页面内容追加到文件
4 测试与代码
4.1 测试
分别测试直接询问与使用 ACT询问，结果如下：
直接搜索
直接搜索
使用 ACT
使用 ACT
使用 ACT
4.2 代码
完整代码与详细注释如下：
# -*- coding: utf-8 -*-
from typing import Tuple # 导入typing模块，用于类型注解
from kOS import nact # 导入kOS系统的nact模块，提供了一系列原生ACT函数
# 定义输入内容解析函数
def handle_search(query: str) -> Tuple[str, ...]:
search_list = [query]
# 创建批处理器
batcher: nact.KOSConcurrentBatcher = nact.k_concurrent_create_batcher()
# 方案一：以递归方式调用上一次重写的结果
'''
count = 0
while count < 3:
if count != 0:
# 如果不是第一次循环，则在函数中加入上下文
search = nact.k_semantic_rephrase('在不影响本意的情况下，以提问的方式重写这个问题', query, f'参考：{search}', '')
else:
# 第一次循环则直接调用函数
search = nact.k_semantic_rephrase('在不影响本意的情况下，以提问的方式重写这个问题', query, '', '')
search_list.append(search)
count += 1
'''
# 方案二：调用两次，分别为缩写与扩写
batcher.submit(nact.k_semantic_rephrase, '在不影响本意的情况下缩写这个问题', query, '', '')
batcher.submit(nact.k_semantic_rephrase, '在不影响本意的情况下，以提问的方式扩写这个问题', query, '', '')
# 等待批处理任务完成
results: list = batcher.wait()
for result in results:
if not result:
continue
search_list.append(result) # 获取批处理任务的结果
return search_list
# 定义处理单个搜索结果主题的函数
def handle_topic(query: str, topic: nact.KOSWebSearchTopic, idx: int) -> Tuple[nact.KOSWebPageData, str, str]:
# 获取网页数据
page_data: nact.KOSWebPageData = nact.k_web_get_page_data(topic.source_url)
if not page_data or not page_data.content:
return None, '', '' # 如果没有获取到数据或内容，返回空值
# 设置标题，如果主题没有标题，则使用网页数据的标题
title = topic.title if topic.title else page_data.title
if not title: # 如果还是没有标题
title = str(nact.k_time_cur_milliseconds()) + str(idx) # 使用当前时间毫秒数和索引作为标题
ref_tag = page_data.domain # 使用页面域名作为引用标签
else:
title = title.replace('/', '') # 替换掉标题中的斜杠
ref_tag = title # 标题作为引用标签
page_data.title = title # 更新网页数据的标题
# 对页面内容进行语义总结，生成摘要
summary = nact.k_semantic_summarize_answer(page_data.content, query)
return page_data, ref_tag, summary # 返回网页数据、引用标签和摘要
# 定义记录搜索结果主题的函数
def record_topic(topic: nact.KOSWebSearchTopic, page_data: nact.KOSWebPageData):
# 获取当前本地日期字符串
time_str = nact.k_time_cur_local_day_str()
if nact.k_file_is_available(): # 检查文件系统是否可用
# 打开文件，准备将内容写入，如果文件存在则覆盖
file = nact.k_except_wrapper(nact.k_file_open, f'/种草/{time_str}_{page_data.title}.txt', nact.FileOpenMode.OVERWRITE,
source=nact.FileSource.WEB_LINK, source_url=topic.source_url)
if file: # 如果成功打开文件
nact.k_file_append(file, page_data.content) # 将页面内容追加到文件
def main():
# 获取ACT触发启动时的消息
query = nact.k_get_act_query()
if not query:
query = nact.k_ask_for_answer('请输入想种草的内容') # 如果没有查询要求，请求用户输入
# 分析查询要求
search_list = handle_search(query)
# 检查分析结果
# nact.k_message_send('分析结果：' + str(search_list))
# 创建批处理器
batcher: nact.KOSConcurrentBatcher = nact.k_concurrent_create_batcher()
all_tl = []
# 遍历问题，执行网络搜索
search_count = 0
for search in search_list:
tl = nact.k_web_search(search)
if not tl: # 没有搜索结果则跳过
search_count += 1
continue
for idx, topic in enumerate(tl): # 有结果则遍历搜索结果
all_tl.append(tl) # 将搜索结果添加到列表
batcher.submit(handle_topic, query, topic, idx) # 提交处理任务到批处理器
# 如果所有搜索都未搜索到结果则结束
if search_count == len(search_list):
nact.k_message_send('未搜索到相关内容！')
return None
# 等待批处理任务完成
results: list = batcher.wait()
# 初始化引用源对象和摘要字符串
ref_source = nact.KOSMsgWebContentRefSource()
all_summary = ''
# 遍历批处理结果
for result in results:
if not result:
continue
(page_data, ref_tag, summary) = result
if not page_data:
continue
# 添加引用网页到引用源对象
ref_source.add_web_ref(ref_tag, page_data.url)
all_summary += summary + "\n" # 将摘要追加到字符串
# 对所有摘要进行语义总结
summary = nact.k_semantic_summarize_answer(all_summary, query)
if not summary:
nact.k_message_send('未搜索总结出有效内容') # 如果没有总结内容，发送消息告知用户
return
# 发送总结后的内容给用户
nact.k_message_send(summary, ref_source)
# 打印日志信息
nact.k_print('开始记录种草结果')
# 再次遍历批处理结果，提交记录任务到批处理器
for idx, result in enumerate(results):
if not result:
continue
(page_data, ref_tag, summary) = result
if not page_data:
continue
batcher.submit(record_topic, all_tl[idx], page_data) # 提交记录主题的任务
# 等待记录任务完成
batcher.wait()
# 调用主函数执行
main()
4 小结
- 代码写完后才最终确定了 ACT 名，叫「种草」。
- 原本是想要收集小红书上的数据再处理，但考虑到法律风险与技术难度，决定还是先从 ACT 默认的搜索入手。
- 在难点的解析输入内容里，原本是想先拆分再多次组合，但尝试了几次后决定还是先以简单实现为主，后续有想法再更新。
- 实际运行速度非常慢，实现的效果和我想的也有差距，甚至说相比直接使用搜一搜几乎没有提升。 因此此 ACT 目前仅作学习用途。或许未来会更新？
