Feature: 为 WebSearch 添加智谱 AI 搜索引擎支持及相关配置 #2812
Closed
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
PaloMiku
changed the title
PaloMiku
changed the title
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey there - I've reviewed your changes - here's some feedback:
- There’s duplicated ZAI HTTP call logic in main.py and engines/zai.py—consolidate into a single wrapper to avoid redundancy.
- The repeated remove_tool calls in edit_web_search_tools could be extracted into a helper or loop to reduce boilerplate and improve readability.
- The change of default max_agent_step from 30 to 10 is orthogonal to ZAI integration—consider separating or justifying that adjustment in its own PR.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- There’s duplicated ZAI HTTP call logic in main.py and engines/zai.py—consolidate into a single wrapper to avoid redundancy.
- The repeated remove_tool calls in edit_web_search_tools could be extracted into a helper or loop to reduce boilerplate and improve readability.
- The change of default max_agent_step from 30 to 10 is orthogonal to ZAI integration—consider separating or justifying that adjustment in its own PR.
## Individual Comments
### Comment 1
<location> `packages/web_searcher/main.py:205-214` </location>
<code_context>
+
+ logger.info(f"ZAI搜索 - 查询: {query} | 引擎: {self.search_engine}")
+
+ async with aiohttp.ClientSession(trust_env=True) as session:
+ try:
+ async with session.post(
+ f"{self.base_url}/web_search",
+ json=payload,
+ headers=headers,
+ timeout=aiohttp.ClientTimeout(total=self.timeout),
+ ) as response:
+ if response.status != 200:
+ error_text = await response.text()
+ raise Exception(
+ f"ZAI API错误: {response.status} - {error_text}"
+ )
+
+ data = await response.json()
+ results = self._parse_search_results(data)
+ logger.info(f"ZAI搜索完成 - 返回 {len(results)} 个结果")
</code_context>
<issue_to_address>
**suggestion (performance):** Timeout value for ZAI search is set to 6 seconds, which may be too short for some network conditions.
The aiohttp timeout here is much shorter than the 30 seconds used elsewhere in the ZAI engine wrapper. Increasing it would help prevent failures on slow networks or with large queries.
</issue_to_address>
### Comment 2
<location> `packages/web_searcher/main.py:413` </location>
<code_context>
+ "websearch_zai_content_size", "high"
+ )
+
+ if search_recency_filter == "oneWeek":
+ search_recency_filter = prov_settings.get(
+ "websearch_zai_recency_filter", "oneWeek"
</code_context>
<issue_to_address>
**issue (bug_risk):** The logic for search_recency_filter may override explicit user input.
This assignment replaces the user's 'oneWeek' input with the config value. To preserve user intent, check if search_recency_filter is None before applying the default.
</issue_to_address>
### Comment 3
<location> `packages/web_searcher/main.py:466-471` </location>
<code_context>
if not websearch_enable:
- # pop tools
for tool_name in self.TOOLS:
- tool_set.remove_tool(tool_name)
+ try:
+ tool_set.remove_tool(tool_name)
+ except Exception:
+ pass
return
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Suppressing all exceptions when removing tools may hide underlying issues.
Instead of passing silently, log the exception or only catch specific expected errors to avoid hiding bugs.
```suggestion
for tool_name in self.TOOLS:
try:
tool_set.remove_tool(tool_name)
except Exception as e:
logger.warning(f"Failed to remove tool '{tool_name}': {e}", exc_info=True)
return
```
</issue_to_address>
### Comment 4
<location> `packages/web_searcher/main.py:449` </location>
<code_context>
+ return ret
+
@filter.on_llm_request(priority=-10000)
async def edit_web_search_tools(
self, event: AstrMessageEvent, req: ProviderRequest
- ) -> str:
</code_context>
<issue_to_address>
**issue (complexity):** Consider centralizing provider-to-tool mapping and key rotation logic to reduce duplication and improve maintainability.
```python
# 1. Consolidate provider→tool mapping at the class level
class Main(star.Star):
TOOLS = {
"default": ["web_search", "fetch_url"],
"tavily": ["web_search_tavily", "tavily_extract_web_page"],
"zai": ["web_search_zai"],
}
ALL_TOOLS = sum(TOOLS.values(), [])
@filter.on_llm_request(priority=-10000)
async def edit_web_search_tools(self, event, req: ProviderRequest):
cfg = self.context.get_config(umo=event.unified_msg_origin)
prov = cfg.get("provider_settings", {})
websearch_enabled = prov.get("web_search", False)
provider = prov.get("websearch_provider", "default")
func_mgr = self.context.get_llm_tool_manager()
# normalize func_tool
tool_set = req.func_tool
if isinstance(tool_set, FunctionToolManager):
tool_set = req.func_tool = tool_set.get_full_tool_set()
if not websearch_enabled:
for t in self.ALL_TOOLS:
tool_set.remove_tool(t, ignore_errors=True)
return
# calculate add/remove lists
keep = set(self.TOOLS.get(provider, self.TOOLS["default"]))
remove = set(self.ALL_TOOLS) - keep
for name in remove:
tool_set.remove_tool(name, ignore_errors=True)
for name in keep:
fn = func_mgr.get_func(name)
if fn:
tool_set.add_tool(fn)
```
```python
# 2. Merge _get_tavily_key and _get_zai_key into one rotation helper
class Main(...):
def __init__(...):
self._key_indexes = defaultdict(int)
self._key_locks = defaultdict(asyncio.Lock)
async def _rotate_key(self, cfg: AstrBotConfig, key_name: str) -> str:
keys = cfg.get("provider_settings", {}).get(key_name, [])
if not keys:
raise ValueError(f"Error: {key_name} is not configured")
lock = self._key_locks[key_name]
async with lock:
idx = self._key_indexes[key_name]
key = keys[idx]
self._key_indexes[key_name] = (idx + 1) % len(keys)
return key
# existing methods now become thin wrappers:
async def _get_tavily_key(self, cfg):
return await self._rotate_key(cfg, "websearch_tavily_key")
async def _get_zai_key(self, cfg):
return await self._rotate_key(cfg, "websearch_zai_keys")
```
These changes collapse duplicated branching in `edit_web_search_tools` into a single loop driven by a mapping, and unify the key‐rotation logic into one helper, preserving all current behavior.
</issue_to_address>
### Comment 5
<location> `packages/web_searcher/engines/zai.py:102-104` </location>
<code_context>
</code_context>
<issue_to_address>
**issue (code-quality):** Raise a specific error instead of the general `Exception` or `BaseException` ([`raise-specific-error`](https://docs.sourcery.ai/Reference/Rules-and-In-Line-Suggestions/Python/Default-Rules/raise-specific-error))
<details><summary>Explanation</summary>If a piece of code raises a specific exception type
rather than the generic
[`BaseException`](https://docs.python.org/3/library/exceptions.html#BaseException)
or [`Exception`](https://docs.python.org/3/library/exceptions.html#Exception),
the calling code can:
- get more information about what type of error it is
- define specific exception handling for it
This way, callers of the code can handle the error appropriately.
How can you solve this?
- Use one of the [built-in exceptions](https://docs.python.org/3/library/exceptions.html) of the standard library.
- [Define your own error class](https://docs.python.org/3/tutorial/errors.html#tut-userexceptions) that subclasses `Exception`.
So instead of having code raising `Exception` or `BaseException` like
```python
if incorrect_input(value):
raise Exception("The input is incorrect")
```
you can have code raising a specific error like
```python
if incorrect_input(value):
raise ValueError("The input is incorrect")
```
or
```python
class IncorrectInputError(Exception):
pass
if incorrect_input(value):
raise IncorrectInputError("The input is incorrect")
```
</details>
</issue_to_address>
### Comment 6
<location> `packages/web_searcher/engines/zai.py:112` </location>
<code_context>
</code_context>
<issue_to_address>
**issue (code-quality):** Raise a specific error instead of the general `Exception` or `BaseException` ([`raise-specific-error`](https://docs.sourcery.ai/Reference/Rules-and-In-Line-Suggestions/Python/Default-Rules/raise-specific-error))
<details><summary>Explanation</summary>If a piece of code raises a specific exception type
rather than the generic
[`BaseException`](https://docs.python.org/3/library/exceptions.html#BaseException)
or [`Exception`](https://docs.python.org/3/library/exceptions.html#Exception),
the calling code can:
- get more information about what type of error it is
- define specific exception handling for it
This way, callers of the code can handle the error appropriately.
How can you solve this?
- Use one of the [built-in exceptions](https://docs.python.org/3/library/exceptions.html) of the standard library.
- [Define your own error class](https://docs.python.org/3/tutorial/errors.html#tut-userexceptions) that subclasses `Exception`.
So instead of having code raising `Exception` or `BaseException` like
```python
if incorrect_input(value):
raise Exception("The input is incorrect")
```
you can have code raising a specific error like
```python
if incorrect_input(value):
raise ValueError("The input is incorrect")
```
or
```python
class IncorrectInputError(Exception):
pass
if incorrect_input(value):
raise IncorrectInputError("The input is incorrect")
```
</details>
</issue_to_address>
### Comment 7
<location> `packages/web_searcher/main.py:211-213` </location>
<code_context>
</code_context>
<issue_to_address>
**issue (code-quality):** Raise a specific error instead of the general `Exception` or `BaseException` ([`raise-specific-error`](https://docs.sourcery.ai/Reference/Rules-and-In-Line-Suggestions/Python/Default-Rules/raise-specific-error))
<details><summary>Explanation</summary>If a piece of code raises a specific exception type
rather than the generic
[`BaseException`](https://docs.python.org/3/library/exceptions.html#BaseException)
or [`Exception`](https://docs.python.org/3/library/exceptions.html#Exception),
the calling code can:
- get more information about what type of error it is
- define specific exception handling for it
This way, callers of the code can handle the error appropriately.
How can you solve this?
- Use one of the [built-in exceptions](https://docs.python.org/3/library/exceptions.html) of the standard library.
- [Define your own error class](https://docs.python.org/3/tutorial/errors.html#tut-userexceptions) that subclasses `Exception`.
So instead of having code raising `Exception` or `BaseException` like
```python
if incorrect_input(value):
raise Exception("The input is incorrect")
```
you can have code raising a specific error like
```python
if incorrect_input(value):
raise ValueError("The input is incorrect")
```
or
```python
class IncorrectInputError(Exception):
pass
if incorrect_input(value):
raise IncorrectInputError("The input is incorrect")
```
</details>
</issue_to_address>
### Comment 8
<location> `packages/web_searcher/engines/zai.py:70` </location>
<code_context>
def _build_payload(
self,
query: str,
count: int,
search_domain_filter: str,
search_recency_filter: str,
content_size: str,
) -> Dict[str, Any]:
"""构建请求参数"""
payload = {
"search_engine": self.search_engine,
"search_query": query,
"count": min(max(count, 1), 50),
}
if search_domain_filter:
payload["search_domain_filter"] = search_domain_filter
if search_recency_filter != "noLimit":
payload["search_recency_filter"] = search_recency_filter
if content_size in ["low", "medium", "high"]:
payload["content_size"] = content_size
return payload
</code_context>
<issue_to_address>
**suggestion (code-quality):** Use set when checking membership of a collection of literals ([`collection-into-set`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/collection-into-set/))
```suggestion
if content_size in {"low", "medium", "high"}:
```
</issue_to_address>
### Comment 9
<location> `packages/web_searcher/engines/zai.py:111-112` </location>
<code_context>
async def search(
self,
query: str,
count: int = 10,
search_domain_filter: str = "",
search_recency_filter: str = "oneWeek",
content_size: str = "high",
) -> List[SearchResult]:
"""使用 Z.AI Web Search API 进行搜索"""
api_key = self._get_random_api_key()
headers = self._build_headers(api_key)
payload = self._build_payload(
query, count, search_domain_filter, search_recency_filter, content_size
)
logger.info(f"ZAI搜索 - 查询: {query} | 引擎: {self.search_engine}")
async with aiohttp.ClientSession(trust_env=True) as session:
try:
async with session.post(
f"{self.base_url}/web_search",
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(total=self.timeout),
) as response:
if response.status != 200:
error_text = await response.text()
raise Exception(
f"ZAI API错误: {response.status} - {error_text}"
)
data = await response.json()
results = self._parse_search_results(data)
logger.info(f"ZAI搜索完成 - 返回 {len(results)} 个结果")
return results
except asyncio.TimeoutError:
raise Exception("ZAI API请求超时")
except Exception as e:
logger.error(f"ZAI搜索失败: {e}")
raise
</code_context>
<issue_to_address>
**suggestion (code-quality):** Explicitly raise from a previous error ([`raise-from-previous-error`](https://docs.sourcery.ai/Reference/Default-Rules/suggestions/raise-from-previous-error/))
```suggestion
except asyncio.TimeoutError as e:
raise Exception("ZAI API请求超时") from e
```
</issue_to_address>
### Comment 10
<location> `packages/web_searcher/main.py:32` </location>
<code_context>
def __init__(self, context: star.Context) -> None:
self.context = context
self.tavily_key_index = 0
self.tavily_key_lock = asyncio.Lock()
self.zai_key_index = 0
self.zai_key_lock = asyncio.Lock()
# 将 str 类型的 key 迁移至 list[str]，并保存
cfg = self.context.get_config()
provider_settings = cfg.get("provider_settings")
if provider_settings:
tavily_key = provider_settings.get("websearch_tavily_key")
if isinstance(tavily_key, str):
logger.info(
"检测到旧版 websearch_tavily_key (字符串格式)，自动迁移为列表格式并保存。"
)
if tavily_key:
provider_settings["websearch_tavily_key"] = [tavily_key]
else:
provider_settings["websearch_tavily_key"] = []
cfg.save_config()
self.bing_search = Bing()
self.sogo_search = Sogo()
self.google = Google()
</code_context>
<issue_to_address>
**issue (code-quality):** We've found these issues:
- Use named expression to simplify assignment and conditional ([`use-named-expression`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-named-expression/))
- Replace if statement with if expression ([`assign-if-exp`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/assign-if-exp/))
</issue_to_address>
### Comment 11
<location> `packages/web_searcher/main.py:504-505` </location>
<code_context>
@filter.on_llm_request(priority=-10000)
async def edit_web_search_tools(
self, event: AstrMessageEvent, req: ProviderRequest
):
"""动态管理搜索工具的可用性"""
cfg = self.context.get_config(umo=event.unified_msg_origin)
prov_settings = cfg.get("provider_settings", {})
websearch_enable = prov_settings.get("web_search", False)
provider = prov_settings.get("websearch_provider", "default")
logger.info(f"web_searcher - provider: {provider}, enabled: {websearch_enable}")
tool_set = req.func_tool
if isinstance(tool_set, FunctionToolManager):
req.func_tool = tool_set.get_full_tool_set()
tool_set = req.func_tool
if not websearch_enable:
for tool_name in self.TOOLS:
try:
tool_set.remove_tool(tool_name)
except Exception:
pass
return
func_tool_mgr = self.context.get_llm_tool_manager()
if provider == "default":
web_search_t = func_tool_mgr.get_func("web_search")
fetch_url_t = func_tool_mgr.get_func("fetch_url")
if web_search_t:
tool_set.add_tool(web_search_t)
if fetch_url_t:
tool_set.add_tool(fetch_url_t)
for tool in [
"web_search_tavily",
"tavily_extract_web_page",
"web_search_zai",
]:
try:
tool_set.remove_tool(tool)
except Exception:
pass
elif provider == "tavily":
web_search_tavily = func_tool_mgr.get_func("web_search_tavily")
tavily_extract_web_page = func_tool_mgr.get_func("tavily_extract_web_page")
if web_search_tavily:
tool_set.add_tool(web_search_tavily)
if tavily_extract_web_page:
tool_set.add_tool(tavily_extract_web_page)
for tool in ["web_search", "fetch_url", "web_search_zai"]:
try:
tool_set.remove_tool(tool)
except Exception:
pass
elif provider == "zai":
web_search_zai = func_tool_mgr.get_func("web_search_zai")
if web_search_zai:
tool_set.add_tool(web_search_zai)
for tool in [
"web_search",
"fetch_url",
"web_search_tavily",
"tavily_extract_web_page",
]:
try:
tool_set.remove_tool(tool)
except Exception:
pass
</code_context>
<issue_to_address>
**suggestion (code-quality):** Use named expression to simplify assignment and conditional ([`use-named-expression`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-named-expression/))
```suggestion
if web_search_zai := func_tool_mgr.get_func("web_search_zai"):
```
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Contributor
Author
|
理论上很好，但是有时候会循环查询某一个问题，瞬间就蒸发大量搜索额度（ |
Contributor
Author
|
Member
|
有没有不需要新增依赖的方案 > < |
Contributor
Author
这不大现实，调用他们搜索 api sdk 的基本是他们依赖包给的 而且现在还有 bug（反复调用，我还没看出来问题在模型还是代码上(~_~💧)） |
Contributor
Author
不需要新增依赖的方案也有，用智谱的 MCP，但是可自定义部分内容很少而且我自己测试就没成功调用过 |
Contributor
Author
换个别人现成的 MCP 倒是成功调用了 反复调用应该是模型的问题（为什么智谱的 GLM 4.5 加自家 AI 搜索会不停调用啊QAQ，自家模型和自家产品适配都做不好吗？换 Moonshot 和 Deepseek 模型都不会有反复调用的问题） |
Contributor
Author
好吧也有但是智谱的最严重 |
PaloMiku
force-pushed
the
feature/zai-search
branch
from
b147274 to
09d1f96
Compare
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
fixes #2811
Motivation / 动机
为 AstrBot WebSearch 增加了智谱搜索 API 的接入支持，优势具体介绍请查看相关 Issue .
Modifications / 改动点
引入了智谱 AI 搜索的 Python 包，以及它的API支持，修改了和增加 WebSearch 部分的组件代码。
Verification Steps / 验证步骤
Clone 并启动代码，在设置中增加智谱 AI （zai）搜索信息，结合 LLM 测试。
Screenshots or Test Results / 运行截图或测试结果
Compatibility & Breaking Changes / 兼容性与破坏性变更
Checklist / 检查清单
requirements.txt和pyproject.toml文件相应位置。/ I have ensured that no new dependencies are introduced, OR if new dependencies are introduced, they have been added to the appropriate locations inrequirements.txtandpyproject.toml.Sourcery 总结
将智谱 AI (ZAI) 作为 AstrBot 的新 WebSearch 提供商，通过添加 SDK 依赖、API 包装器、配置选项、动态工具注册以及配套文档。
新功能：
改进：
构建：
文档：
杂项：
Original summary in English
Summary by Sourcery
Integrate Zhipu AI (ZAI) as a new WebSearch provider in AstrBot by adding SDK dependency, API wrapper, configuration options, dynamic tool registration, and accompanying documentation.
New Features:
Enhancements:
Build:
Documentation:
Chores: