gardel/Mofox-Core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

generator修改与文档

Browse Source
This commit is contained in:
UnCLAS-Prommer
2025-07-27 16:59:33 +08:00
parent 5470f68f4a
commit 96d7ad527a
4 changed files with 243 additions and 366 deletions
Download Patch File Download Diff File Expand all files Collapse all files

403
docs/plugins/api/generator-api.md
View File

@@ -6,241 +6,150 @@
```python
from src.plugin_system.apis import generator_api
# 或者
from src.plugin_system import generator_api
```
## 主要功能
### 1. 回复器获取
#### `get_replyer(chat_stream=None, platform=None, chat_id=None, is_group=True)`
```python
def get_replyer(
chat_stream: Optional[ChatStream] = None,
chat_id: Optional[str] = None,
model_configs: Optional[List[Dict[str, Any]]] = None,
request_type: str = "replyer",
) -> Optional[DefaultReplyer]:
```
获取回复器对象
**参数：**
- `chat_stream`：聊天流对象（优先）
- `platform`：平台名称，如"qq"
- `chat_id`：聊天ID（群ID或用户ID）
- `is_group`：是否为群聊
优先使用chat_stream，如果没有则使用chat_id直接查找。
**返回：**
- `DefaultReplyer`：回复器对象，如果获取失败则返回None
使用 ReplyerManager 来管理实例，避免重复创建。
**示例：**
**Args:**
- `chat_stream`: 聊天流对象
- `chat_id`: 聊天ID（实际上就是`stream_id`）
- `model_configs`: 模型配置
- `request_type`: 请求类型，用于记录LLM使用情况，可以不写
**Returns:**
- `DefaultReplyer`: 回复器对象，如果获取失败则返回None
#### 示例
```python
# 使用聊天流获取回复器
replyer = generator_api.get_replyer(chat_stream=chat_stream)
# 使用平台和ID获取回复器
replyer = generator_api.get_replyer(
platform="qq",
chat_id="123456789",
is_group=True
)
replyer = generator_api.get_replyer(chat_id="123456789")
```
### 2. 回复生成
#### `generate_reply(chat_stream=None, action_data=None, platform=None, chat_id=None, is_group=True)`
```python
async def generate_reply(
chat_stream: Optional[ChatStream] = None,
chat_id: Optional[str] = None,
action_data: Optional[Dict[str, Any]] = None,
reply_to: str = "",
extra_info: str = "",
available_actions: Optional[Dict[str, ActionInfo]] = None,
enable_tool: bool = False,
enable_splitter: bool = True,
enable_chinese_typo: bool = True,
return_prompt: bool = False,
model_configs: Optional[List[Dict[str, Any]]] = None,
request_type: str = "",
) -> Tuple[bool, List[Tuple[str, Any]], Optional[str]]:
```
生成回复
**参数：**
- `chat_stream`：聊天流对象（优先）
- `action_data`：动作数据
- `platform`：平台名称（备用）
- `chat_id`：聊天ID（备用）
- `is_group`：是否为群聊（备用）
优先使用chat_stream，如果没有则使用chat_id直接查找。
**返回：**
- `Tuple[bool, List[Tuple[str, Any]]]`：(是否成功, 回复集合)
**Args:**
- `chat_stream`: 聊天流对象
- `chat_id`: 聊天ID（实际上就是`stream_id`）
- `action_data`: 动作数据（向下兼容，包含`reply_to`和`extra_info`）
- `reply_to`: 回复目标，格式为 `{发送者的person_name:消息内容}`
- `extra_info`: 附加信息
- `available_actions`: 可用动作字典，格式为 `{"action_name": ActionInfo}`
- `enable_tool`: 是否启用工具
- `enable_splitter`: 是否启用分割器
- `enable_chinese_typo`: 是否启用中文错别字
- `return_prompt`: 是否返回提示词
- `model_configs`: 模型配置，可选
- `request_type`: 请求类型，用于记录LLM使用情况
**示例：**
**Returns:**
- `Tuple[bool, List[Tuple[str, Any]], Optional[str]]`: (是否成功, 回复集合, 提示词)
#### 示例
```python
success, reply_set = await generator_api.generate_reply(
success, reply_set, prompt = await generator_api.generate_reply(
chat_stream=chat_stream,
action_data={"message": "你好", "intent": "greeting"}
action_data=action_data,
reply_to="麦麦:你好",
available_actions=action_info,
enable_tool=True,
return_prompt=True
)
if success:
for reply_type, reply_content in reply_set:
print(f"回复类型: {reply_type}, 内容: {reply_content}")
if prompt:
print(f"使用的提示词: {prompt}")
```
#### `rewrite_reply(chat_stream=None, reply_data=None, platform=None, chat_id=None, is_group=True)`
重写回复
**参数：**
- `chat_stream`：聊天流对象（优先）
- `reply_data`：回复数据
- `platform`：平台名称（备用）
- `chat_id`：聊天ID（备用）
- `is_group`：是否为群聊（备用）
**返回：**
- `Tuple[bool, List[Tuple[str, Any]]]`：(是否成功, 回复集合)
**示例：**
### 3. 回复重写
```python
success, reply_set = await generator_api.rewrite_reply(
async def rewrite_reply(
chat_stream: Optional[ChatStream] = None,
reply_data: Optional[Dict[str, Any]] = None,
chat_id: Optional[str] = None,
enable_splitter: bool = True,
enable_chinese_typo: bool = True,
model_configs: Optional[List[Dict[str, Any]]] = None,
raw_reply: str = "",
reason: str = "",
reply_to: str = "",
return_prompt: bool = False,
) -> Tuple[bool, List[Tuple[str, Any]], Optional[str]]:
```
重写回复，使用新的内容替换旧的回复内容。
优先使用chat_stream，如果没有则使用chat_id直接查找。
**Args:**
- `chat_stream`: 聊天流对象
- `reply_data`: 回复数据，包含`raw_reply`, `reason`和`reply_to`，**（向下兼容备用，当其他参数缺失时从此获取）**
- `chat_id`: 聊天ID（实际上就是`stream_id`）
- `enable_splitter`: 是否启用分割器
- `enable_chinese_typo`: 是否启用中文错别字
- `model_configs`: 模型配置，可选
- `raw_reply`: 原始回复内容
- `reason`: 重写原因
- `reply_to`: 回复目标，格式为 `{发送者的person_name:消息内容}`
**Returns:**
- `Tuple[bool, List[Tuple[str, Any]], Optional[str]]`: (是否成功, 回复集合, 提示词)
#### 示例
```python
success, reply_set, prompt = await generator_api.rewrite_reply(
chat_stream=chat_stream,
reply_data={"original_text": "原始回复", "style": "more_friendly"}
raw_reply="原始回复内容",
reason="重写原因",
reply_to="麦麦:你好",
return_prompt=True
)
if success:
for reply_type, reply_content in reply_set:
print(f"回复类型: {reply_type}, 内容: {reply_content}")
if prompt:
print(f"使用的提示词: {prompt}")
```
## 使用示例
### 1. 基础回复生成
```python
from src.plugin_system.apis import generator_api
async def generate_greeting_reply(chat_stream, user_name):
"""生成问候回复"""
action_data = {
"intent": "greeting",
"user_name": user_name,
"context": "morning_greeting"
}
success, reply_set = await generator_api.generate_reply(
chat_stream=chat_stream,
action_data=action_data
)
if success and reply_set:
# 获取第一个回复
reply_type, reply_content = reply_set[0]
return reply_content
return "你好！" # 默认回复
```
### 2. 在Action中使用回复生成器
```python
from src.plugin_system.base import BaseAction
class ChatAction(BaseAction):
async def execute(self, action_data, chat_stream):
# 准备回复数据
reply_context = {
"message_type": "response",
"user_input": action_data.get("user_message", ""),
"intent": action_data.get("intent", ""),
"entities": action_data.get("entities", {}),
"context": self.get_conversation_context(chat_stream)
}
# 生成回复
success, reply_set = await generator_api.generate_reply(
chat_stream=chat_stream,
action_data=reply_context
)
if success:
return {
"success": True,
"replies": reply_set,
"generated_count": len(reply_set)
}
return {
"success": False,
"error": "回复生成失败",
"fallback_reply": "抱歉，我现在无法理解您的消息。"
}
```
### 3. 多样化回复生成
```python
async def generate_diverse_replies(chat_stream, topic, count=3):
"""生成多个不同风格的回复"""
styles = ["formal", "casual", "humorous"]
all_replies = []
for i, style in enumerate(styles[:count]):
action_data = {
"topic": topic,
"style": style,
"variation": i
}
success, reply_set = await generator_api.generate_reply(
chat_stream=chat_stream,
action_data=action_data
)
if success and reply_set:
all_replies.extend(reply_set)
return all_replies
```
### 4. 回复重写功能
```python
async def improve_reply(chat_stream, original_reply, improvement_type="more_friendly"):
"""改进原始回复"""
reply_data = {
"original_text": original_reply,
"improvement_type": improvement_type,
"target_audience": "young_users",
"tone": "positive"
}
success, improved_replies = await generator_api.rewrite_reply(
chat_stream=chat_stream,
reply_data=reply_data
)
if success and improved_replies:
# 返回改进后的第一个回复
_, improved_content = improved_replies[0]
return improved_content
return original_reply # 如果改进失败，返回原始回复
```
### 5. 条件回复生成
```python
async def conditional_reply_generation(chat_stream, user_message, user_emotion):
"""根据用户情感生成条件回复"""
# 根据情感调整回复策略
if user_emotion == "sad":
action_data = {
"intent": "comfort",
"tone": "empathetic",
"style": "supportive"
}
elif user_emotion == "angry":
action_data = {
"intent": "calm",
"tone": "peaceful",
"style": "understanding"
}
else:
action_data = {
"intent": "respond",
"tone": "neutral",
"style": "helpful"
}
action_data["user_message"] = user_message
action_data["user_emotion"] = user_emotion
success, reply_set = await generator_api.generate_reply(
chat_stream=chat_stream,
action_data=action_data
)
return reply_set if success else []
```
## 回复集合格式
## 回复集合`reply_set`格式
### 回复类型
生成的回复集合包含多种类型的回复：
@@ -260,82 +169,32 @@ reply_set = [
]
```
## 高级用法
### 1. 自定义回复器配置
### 4. 自定义提示词回复
```python
async def generate_with_custom_config(chat_stream, action_data):
"""使用自定义配置生成回复"""
# 获取回复器
replyer = generator_api.get_replyer(chat_stream=chat_stream)
if replyer:
# 可以访问回复器的内部方法
success, reply_set = await replyer.generate_reply_with_context(
reply_data=action_data,
# 可以传递额外的配置参数
)
return success, reply_set
return False, []
async def generate_response_custom(
chat_stream: Optional[ChatStream] = None,
chat_id: Optional[str] = None,
model_configs: Optional[List[Dict[str, Any]]] = None,
prompt: str = "",
) -> Optional[str]:
```
生成自定义提示词回复
### 2. 回复质量评估
优先使用chat_stream，如果没有则使用chat_id直接查找。
```python
async def generate_and_evaluate_replies(chat_stream, action_data):
"""生成回复并评估质量"""
**Args:**
- `chat_stream`: 聊天流对象
- `chat_id`: 聊天ID（备用）
- `model_configs`: 模型配置列表
- `prompt`: 自定义提示词
success, reply_set = await generator_api.generate_reply(
chat_stream=chat_stream,
action_data=action_data
)
if success:
evaluated_replies = []
for reply_type, reply_content in reply_set:
# 简单的质量评估
quality_score = evaluate_reply_quality(reply_content)
evaluated_replies.append({
"type": reply_type,
"content": reply_content,
"quality": quality_score
})
# 按质量排序
evaluated_replies.sort(key=lambda x: x["quality"], reverse=True)
return evaluated_replies
return []
def evaluate_reply_quality(reply_content):
"""简单的回复质量评估"""
if not reply_content:
return 0
score = 50 # 基础分
# 长度适中加分
if 5 <= len(reply_content) <= 100:
score += 20
# 包含积极词汇加分
positive_words = ["好", "棒", "不错", "感谢", "开心"]
for word in positive_words:
if word in reply_content:
score += 10
break
return min(score, 100)
```
**Returns:**
- `Optional[str]`: 生成的自定义回复内容，如果生成失败则返回None
## 注意事项
1. **异步操作**：所有生成函数都是异步的，必须使用`await`
2. **错误处理**：函数内置错误处理，失败时返回False和空列表
3. **聊天流依赖**：需要有效的聊天流对象才能正常工作
4. **性能考虑**：回复生成可能需要一些时间，特别是使用LLM时
5. **回复格式**：返回的回复集合是元组列表，包含类型和内容
6. **上下文感知**：生成器会考虑聊天上下文和历史消息
1. **异步操作**：部分函数是异步的，须使用`await`
2. **聊天流依赖**：需要有效的聊天流对象才能正常工作
3. **性能考虑**：回复生成可能需要一些时间，特别是使用LLM时
4. **回复格式**：返回的回复集合是元组列表，包含类型和内容
5. **上下文感知**：生成器会考虑聊天上下文和历史消息，除非你用的是自定义提示词。

109
src/chat/replyer/default_generator.py
View File

@@ -155,7 +155,6 @@ class DefaultReplyer:
extra_info: str = "",
available_actions: Optional[Dict[str, ActionInfo]] = None,
enable_tool: bool = True,
enable_timeout: bool = False,
) -> Tuple[bool, Optional[str], Optional[str]]:
"""
回复器 (Replier): 负责生成回复文本的核心逻辑。
@@ -165,7 +164,6 @@ class DefaultReplyer:
extra_info: 额外信息，用于补充上下文
available_actions: 可用的动作信息字典
enable_tool: 是否启用工具调用
enable_timeout: 是否启用超时处理
Returns:
Tuple[bool, Optional[str], Optional[str]]: (是否成功, 生成的回复内容, 使用的prompt)
@@ -177,10 +175,9 @@ class DefaultReplyer:
# 3. 构建 Prompt
with Timer("构建Prompt", {}): # 内部计时器，可选保留
prompt = await self.build_prompt_reply_context(
reply_to = reply_to,
reply_to=reply_to,
extra_info=extra_info,
available_actions=available_actions,
enable_timeout=enable_timeout,
enable_tool=enable_tool,
)
@@ -190,29 +187,12 @@ class DefaultReplyer:
# 4. 调用 LLM 生成回复
content = None
reasoning_content = None
model_name = "unknown_model"
# TODO: 复活这里
# reasoning_content = None
# model_name = "unknown_model"
try:
with Timer("LLM生成", {}): # 内部计时器，可选保留
# 加权随机选择一个模型配置
selected_model_config = self._select_weighted_model_config()
logger.info(
f"使用模型生成回复: {selected_model_config.get('name', 'N/A')} (选中概率: {selected_model_config.get('weight', 1.0)})"
)
express_model = LLMRequest(
model=selected_model_config,
request_type=self.request_type,
)
if global_config.debug.show_prompt:
logger.info(f"\n{prompt}\n")
else:
logger.debug(f"\n{prompt}\n")
content, (reasoning_content, model_name) = await express_model.generate_response_async(prompt)
content = await self.llm_generate_content(prompt)
logger.debug(f"replyer生成内容: {content}")
except Exception as llm_e:
@@ -232,7 +212,8 @@ class DefaultReplyer:
raw_reply: str = "",
reason: str = "",
reply_to: str = "",
) -> Tuple[bool, Optional[str]]:
return_prompt: bool = False,
) -> Tuple[bool, Optional[str], Optional[str]]:
"""
表达器 (Expressor): 负责重写和优化回复文本。
@@ -246,8 +227,6 @@ class DefaultReplyer:
Tuple[bool, Optional[str]]: (是否成功, 重写后的回复内容)
"""
try:
with Timer("构建Prompt", {}): # 内部计时器，可选保留
prompt = await self.build_prompt_rewrite_context(
raw_reply=raw_reply,
@@ -256,40 +235,28 @@ class DefaultReplyer:
)
content = None
reasoning_content = None
model_name = "unknown_model"
# TODO: 复活这里
# reasoning_content = None
# model_name = "unknown_model"
if not prompt:
logger.error("Prompt 构建失败，无法生成回复。")
return False, None
return False, None, None
try:
with Timer("LLM生成", {}): # 内部计时器，可选保留
# 加权随机选择一个模型配置
selected_model_config = self._select_weighted_model_config()
logger.info(
f"使用模型重写回复: {selected_model_config.get('name', 'N/A')} (选中概率: {selected_model_config.get('weight', 1.0)})"
)
express_model = LLMRequest(
model=selected_model_config,
request_type=self.request_type,
)
content, (reasoning_content, model_name) = await express_model.generate_response_async(prompt)
content = await self.llm_generate_content(prompt)
logger.info(f"想要表达：{raw_reply}||理由：{reason}||生成回复: {content}\n")
except Exception as llm_e:
# 精简报错信息
logger.error(f"LLM 生成失败: {llm_e}")
return False, None # LLM 调用失败则无法生成回复
return False, None, prompt if return_prompt else None # LLM 调用失败则无法生成回复
return True, content
return True, content, prompt if return_prompt else None
except Exception as e:
logger.error(f"回复生成意外失败: {e}")
traceback.print_exc()
return False, None
return False, None, prompt if return_prompt else None
async def build_relation_info(self, reply_to: str = ""):
if not global_config.relationship.enable_relationship:
@@ -366,9 +333,7 @@ class DefaultReplyer:
if style_habits_str.strip() and grammar_habits_str.strip():
expression_habits_title = "你可以参考以下的语言习惯和句法，如果情景合适就使用，不要盲目使用,不要生硬使用，以合理的方式结合到你的回复中："
expression_habits_block = f"{expression_habits_title}\n{expression_habits_block}"
return expression_habits_block
return f"{expression_habits_title}\n{expression_habits_block}"
async def build_memory_block(self, chat_history: str, target: str) -> str:
"""构建记忆块
@@ -537,7 +502,9 @@ class DefaultReplyer:
duration = end_time - start_time
return name, result, duration
def build_s4u_chat_history_prompts(self, message_list_before_now: List[Dict[str, Any]], target_user_id: str) -> Tuple[str, str]:
def build_s4u_chat_history_prompts(
self, message_list_before_now: List[Dict[str, Any]], target_user_id: str
) -> Tuple[str, str]:
"""
构建 s4u 风格的分离对话 prompt
@@ -647,19 +614,17 @@ class DefaultReplyer:
reply_to: str,
extra_info: str = "",
available_actions: Optional[Dict[str, ActionInfo]] = None,
enable_timeout: bool = False,
enable_tool: bool = True,
) -> str: # sourcery skip: merge-else-if-into-elif, remove-redundant-if
"""
构建回复器上下文
Args:
reply_data: 回复数据
replay_data 包含以下字段：
structured_info: 结构化信息，一般是工具调用获得的信息
reply_to: 回复对象
extra_info/extra_info_block: 额外信息
reply_to: 回复对象，格式为 "发送者:消息内容"
extra_info: 额外信息，用于补充上下文
available_actions: 可用动作
enable_timeout: 是否启用超时处理
enable_tool: 是否启用工具调用
Returns:
str: 构建好的上下文
@@ -1011,6 +976,30 @@ class DefaultReplyer:
display_message=display_message,
)
async def llm_generate_content(self, prompt: str) -> str:
with Timer("LLM生成", {}): # 内部计时器，可选保留
# 加权随机选择一个模型配置
selected_model_config = self._select_weighted_model_config()
logger.info(
f"使用模型生成回复: {selected_model_config.get('name', 'N/A')} (选中概率: {selected_model_config.get('weight', 1.0)})"
)
express_model = LLMRequest(
model=selected_model_config,
request_type=self.request_type,
)
if global_config.debug.show_prompt:
logger.info(f"\n{prompt}\n")
else:
logger.debug(f"\n{prompt}\n")
# TODO: 这里的_应该做出替换
content, _ = await express_model.generate_response_async(prompt)
logger.debug(f"replyer生成内容: {content}")
return content
def weighted_sample_no_replacement(items, weights, k) -> list:
"""
@@ -1069,9 +1058,7 @@ async def get_prompt_info(message: str, threshold: float):
logger.debug(f"获取知识库内容耗时: {(end_time - start_time):.3f}秒")
logger.debug(f"获取知识库内容，相关信息：{related_info[:100]}...，信息长度: {len(related_info)}")
# 格式化知识信息
formatted_prompt_info = f"你有以下这些**知识**：\n{related_info}\n请你**记住上面的知识**，之后可能会用到。\n"
return formatted_prompt_info
return f"你有以下这些**知识**：\n{related_info}\n请你**记住上面的知识**，之后可能会用到。\n"
else:
logger.debug("从LPMM知识库获取知识失败，可能是从未导入过知识，返回空知识...")
return ""

2
src/llm_models/utils_model.py
View File

@@ -851,7 +851,7 @@ class LLMRequest:
def _default_response_handler(
self, result: dict, user_id: str = "system", request_type: str = None, endpoint: str = "/chat/completions"
) -> Tuple:
):
"""默认响应解析"""
if "choices" in result and result["choices"]:
message = result["choices"][0]["message"]

51
src/plugin_system/apis/generator_api.py
View File

@@ -84,18 +84,23 @@ async def generate_reply(
enable_chinese_typo: bool = True,
return_prompt: bool = False,
model_configs: Optional[List[Dict[str, Any]]] = None,
request_type: str = "",
enable_timeout: bool = False,
request_type: str = "generator_api",
) -> Tuple[bool, List[Tuple[str, Any]], Optional[str]]:
"""生成回复
Args:
chat_stream: 聊天流对象（优先）
chat_id: 聊天ID（备用）
action_data: 动作数据
action_data: 动作数据（向下兼容，包含reply_to和extra_info）
reply_to: 回复对象，格式为 "发送者:消息内容"
extra_info: 额外信息，用于补充上下文
available_actions: 可用动作
enable_tool: 是否启用工具调用
enable_splitter: 是否启用消息分割器
enable_chinese_typo: 是否启用错字生成器
return_prompt: 是否返回提示词
model_configs: 模型配置列表
request_type: 请求类型（可选，记录LLM使用）
Returns:
Tuple[bool, List[Tuple[str, Any]], Optional[str]]: (是否成功, 回复集合, 提示词)
"""
@@ -118,7 +123,6 @@ async def generate_reply(
reply_to=reply_to,
extra_info=extra_info,
available_actions=available_actions,
enable_timeout=enable_timeout,
enable_tool=enable_tool,
)
reply_set = []
@@ -154,12 +158,13 @@ async def rewrite_reply(
raw_reply: str = "",
reason: str = "",
reply_to: str = "",
) -> Tuple[bool, List[Tuple[str, Any]]]:
return_prompt: bool = False,
) -> Tuple[bool, List[Tuple[str, Any]], Optional[str]]:
"""重写回复
Args:
chat_stream: 聊天流对象（优先）
reply_data: 回复数据字典（备用，当其他参数缺失时从此获取）
reply_data: 回复数据字典（向下兼容备用，当其他参数缺失时从此获取）
chat_id: 聊天ID（备用）
enable_splitter: 是否启用消息分割器
enable_chinese_typo: 是否启用错字生成器
@@ -167,6 +172,7 @@ async def rewrite_reply(
raw_reply: 原始回复内容
reason: 回复原因
reply_to: 回复对象
return_prompt: 是否返回提示词
Returns:
Tuple[bool, List[Tuple[str, Any]]]: (是否成功, 回复集合)
@@ -176,7 +182,7 @@ async def rewrite_reply(
replyer = get_replyer(chat_stream, chat_id, model_configs=model_configs)
if not replyer:
logger.error("[GeneratorAPI] 无法获取回复器")
return False, []
return False, [], None
logger.info("[GeneratorAPI] 开始重写回复")
@@ -187,10 +193,11 @@ async def rewrite_reply(
reply_to = reply_to or reply_data.get("reply_to", "")
# 调用回复器重写回复
success, content = await replyer.rewrite_reply_with_context(
success, content, prompt = await replyer.rewrite_reply_with_context(
raw_reply=raw_reply,
reason=reason,
reply_to=reply_to,
return_prompt=return_prompt,
)
reply_set = []
if content:
@@ -201,14 +208,14 @@ async def rewrite_reply(
else:
logger.warning("[GeneratorAPI] 重写回复失败")
return success, reply_set
return success, reply_set, prompt if return_prompt else None
except ValueError as ve:
raise ve
except Exception as e:
logger.error(f"[GeneratorAPI] 重写回复时出错: {e}")
return False, []
return False, [], None
async def process_human_text(content: str, enable_splitter: bool, enable_chinese_typo: bool) -> List[Tuple[str, Any]]:
@@ -234,3 +241,27 @@ async def process_human_text(content: str, enable_splitter: bool, enable_chinese
except Exception as e:
logger.error(f"[GeneratorAPI] 处理人形文本时出错: {e}")
return []
async def generate_response_custom(
chat_stream: Optional[ChatStream] = None,
chat_id: Optional[str] = None,
model_configs: Optional[List[Dict[str, Any]]] = None,
prompt: str = "",
) -> Optional[str]:
replyer = get_replyer(chat_stream, chat_id, model_configs=model_configs)
if not replyer:
logger.error("[GeneratorAPI] 无法获取回复器")
return None
try:
logger.debug("[GeneratorAPI] 开始生成自定义回复")
response = await replyer.llm_generate_content(prompt)
if response:
logger.debug("[GeneratorAPI] 自定义回复生成成功")
return response
else:
logger.warning("[GeneratorAPI] 自定义回复生成失败")
return None
except Exception as e:
logger.error(f"[GeneratorAPI] 生成自定义回复时出错: {e}")
return None