chat

feishu.robot.chat

handle_chat

Python

handle_chat(request: dict, max_num_messages: int | None = ..., max_message_length: int | None = ..., stop_words: set[str] | None = ..., stream: bool = ...) -> dict

Python

handle_chat(request: dict, max_num_messages: int | None = ..., max_message_length: int | None = ..., stream: bool = ...) -> dict

Python

handle_chat(request: NestedDict, max_num_messages: int | None = None, max_message_length: int | None = None, stop_words: set[str] | None = None, stream: bool = True) -> dict

回复飞书消息。

这个函数会根据消息的类型选择不同的回复方式。

对于私聊，会访问所有聊天记录，直到遇到停用词，或者达到最大消息数量。 对于群聊，会访问消息及其回复链中所有的消息，直到达到最大消息数量。

Parameters:

Name	Type	Description	Default
request	NestedDict	飞书请求。	required
max_num_messages	int | None	最大消息数量。默认为 feishu.variables 中的 MAX_NUM_MESSAGES。	None
max_message_length	int | None	最大消息长度。默认为 feishu.variables 中的 MAX_MESSAGE_LENGTH。	None
stop_words	set[str] | None	停用词集合。默认为 feishu.variables 中的 STOP_WORDS。	None
stream	bool	是否使用流式 API。默认为 True。	True

功能	实现函数
回复私聊消息	feishu.robot.chat.handle_chat_history
回复群聊消息	feishu.robot.chat.handle_chat_chain

Source code in feishu/robot/chat.py

def handle_chat(
    request: NestedDict,
    max_num_messages: int | None = None,
    max_message_length: int | None = None,
    stop_words: set[str] | None = None,
    stream: bool = True,
) -> dict:
    r"""
    回复飞书消息。

    这个函数会根据消息的类型选择不同的回复方式。

    对于私聊，会访问所有聊天记录，直到遇到停用词，或者达到最大消息数量。
    对于群聊，会访问消息及其回复链中所有的消息，直到达到最大消息数量。

    Args:
        request: 飞书请求。
        max_num_messages: 最大消息数量。默认为 [feishu.variables][] 中的 `MAX_NUM_MESSAGES`。
        max_message_length: 最大消息长度。默认为 [feishu.variables][] 中的 `MAX_MESSAGE_LENGTH`。
        stop_words: 停用词集合。默认为 [feishu.variables][] 中的 `STOP_WORDS`。
        stream: 是否使用流式 API。默认为 `True`。

    | 功能     | 实现函数                                      |
    |--------|-------------------------------------------|
    | 回复私聊消息 | [feishu.robot.chat.handle_chat_history][] |
    | 回复群聊消息 | [feishu.robot.chat.handle_chat_chain][]   |
    """
    if not isinstance(request, NestedDict):
        request = NestedDict(request)
    chat_type = request.event.message.chat_type
    if chat_type in {
        "p2p",
    }:
        return handle_chat_history(request, max_num_messages, max_message_length, stop_words, stream)
    if chat_type in {"group", "meeting", "private", "public"}:
        return handle_chat_chain(request, max_num_messages, max_message_length, stream)
    return {"status": "error", "message": "Unsupported chat type"}

handle_chat_chain

Python

handle_chat_chain(request: dict, max_num_messages: int | None = None, max_message_length: int | None = None, stream: bool = True) -> dict

回复飞书消息及其回复链中所有的消息。

Parameters:

Name	Type	Description	Default
request	dict	飞书请求。	required
max_num_messages	int | None	最大消息数量。默认为 feishu.variables 中的 MAX_NUM_MESSAGES。	None
max_message_length	int | None	最大消息长度。默认为 feishu.variables 中的 MAX_MESSAGE_LENGTH。	None
stream	bool	是否使用流式 API。默认为 True。	True

相关函数

feishu.im.messages.get_messages_chain

Source code in feishu/robot/chat.py

def handle_chat_chain(
    request: NestedDict,
    max_num_messages: int | None = None,
    max_message_length: int | None = None,
    stream: bool = True,
) -> dict:
    r"""
    回复飞书消息及其回复链中所有的消息。

    Args:
        request: 飞书请求。
        max_num_messages: 最大消息数量。默认为 [feishu.variables][] 中的 `MAX_NUM_MESSAGES`。
        max_message_length: 最大消息长度。默认为 [feishu.variables][] 中的 `MAX_MESSAGE_LENGTH`。
        stream: 是否使用流式 API。默认为 `True`。

    相关函数:
        [feishu.im.messages.get_messages_chain][]
    """
    if not isinstance(request, NestedDict):
        request = NestedDict(request)
    message_id = request.event.message.message_id
    max_num_messages = max_num_messages or variables.MAX_NUM_MESSAGES
    max_message_length = max_message_length or variables.MAX_MESSAGE_LENGTH
    try:
        messages = get_messages(message_id, max_num_messages=max_num_messages, max_message_length=max_message_length)
    except FeishuException as e:
        logging.error("Failed to get Feishu messages: %s", e)
        return {"status": "error", "message": "Failed to get Feishu messages"}
    return _handle_chat(messages, message_id, uuid=request.header.event_id, stream=stream)

handle_chat_history

Python

handle_chat_history(request: dict, max_num_messages: int | None = None, max_message_length: int | None = None, stop_words: set[str] | None = None, stream: bool = True) -> dict

回复飞书对话中所有的消息。

Parameters:

Name	Type	Description	Default
request	dict	飞书请求。	required
max_num_messages	int | None	最大消息数量。默认为 feishu.variables 中的 MAX_NUM_MESSAGES。	None
max_message_length	int | None	最大消息长度。默认为 feishu.variables 中的 MAX_MESSAGE_LENGTH。	None
stop_words	set[str] | None	停用词集合。默认为 feishu.variables 中的 STOP_WORDS。	None
stream	bool	是否使用流式 API。默认为 True。	True

相关函数

feishu.im.messages.get_messages_history

Source code in feishu/robot/chat.py

def handle_chat_history(
    request: NestedDict,
    max_num_messages: int | None = None,
    max_message_length: int | None = None,
    stop_words: set[str] | None = None,
    stream: bool = True,
) -> dict:
    r"""
    回复飞书对话中所有的消息。

    Args:
        request: 飞书请求。
        max_num_messages: 最大消息数量。默认为 [feishu.variables][] 中的 `MAX_NUM_MESSAGES`。
        max_message_length: 最大消息长度。默认为 [feishu.variables][] 中的 `MAX_MESSAGE_LENGTH`。
        stop_words: 停用词集合。默认为 [feishu.variables][] 中的 `STOP_WORDS`。
        stream: 是否使用流式 API。默认为 `True`。

    相关函数:
        [feishu.im.messages.get_messages_history][]
    """
    if not isinstance(request, NestedDict):
        request = NestedDict(request)
    stop_words = stop_words or variables.STOP_WORDS
    sender_id = request.event.sender.sender_id.open_id
    chat_id = request.event.message.chat_id
    max_num_messages = max_num_messages or variables.MAX_NUM_MESSAGES
    max_message_length = max_message_length or variables.MAX_MESSAGE_LENGTH
    try:
        messages = get_messages(
            chat_id, max_num_items=max_num_messages, early_stop=partial(early_stop, stop_words=stop_words)
        )
    except FeishuException as e:
        logging.error("Failed to get Feishu messages: %s", e)
        return {"status": "error", "message": "Failed to get Feishu messages"}
    return _handle_chat(messages, sender_id, uuid=request.header.event_id, stream=stream)