vrr/deer-flow
2
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-07-09 16:08:31 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

fix: batch Feishu file messages into one thread (#3753)

Browse source 
* fix: batch feishu file messages

* fix: narrow Feishu file batching
This commit is contained in:
Huixin615 2026-07-04 23:06:28 +08:00 • committed by GitHub
parent dcb2e687d5
commit c22c955c2d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 297 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

136
backend/app/channels/feishu.py
View file

@ -28,6 +28,7 @@ from deerflow.sandbox.sandbox_provider import get_sandbox_provider
logger = logging.getLogger(__name__)
PENDING_CLARIFICATION_TTL_SECONDS = 30 * 60
FEISHU_INBOUND_BATCH_WINDOW_SECONDS = 0.75
def _is_feishu_command(text: str) -> bool:
@ -66,6 +67,7 @@ class FeishuChannel(Channel):
self._running_card_ids: dict[str, str] = {}
self._running_card_tasks: dict[str, asyncio.Task] = {}
self._pending_clarifications: dict[tuple[str, str], list[dict[str, Any]]] = {}
self._pending_inbound_batches: dict[tuple[str, str], dict[str, Any]] = {}
self._CreateFileRequest = None
self._CreateFileRequestBody = None
self._CreateImageRequest = None
@ -715,6 +717,113 @@ class FeishuChannel(Channel):
return root_id or msg_id, False
@staticmethod
def _is_batchable_file_inbound(
*,
msg_type: InboundMessageType,
text: str,
files: list[dict[str, Any]],
root_id: str | None,
parent_id: str | None,
thread_id: str | None,
) -> bool:
return msg_type == InboundMessageType.CHAT and text in {"[file]", "[image]"} and len(files) == 1 and not (root_id or parent_id or thread_id)
def _schedule_prepare_inbound(
self,
msg_id: str,
inbound: InboundMessage,
*,
source_message_ids: list[str] | None = None,
) -> None:
if self._main_loop and self._main_loop.is_running():
logger.info("[Feishu] publishing inbound message to bus (type=%s, msg_id=%s)", inbound.msg_type.value, msg_id)
fut = asyncio.run_coroutine_threadsafe(
self._prepare_inbound(msg_id, inbound, source_message_ids=source_message_ids),
self._main_loop,
)
fut.add_done_callback(lambda f, mid=msg_id: self._log_future_error(f, "prepare_inbound", mid))
else:
logger.warning("[Feishu] main loop not running, cannot publish inbound message")
def _schedule_batch_flush(self, key: tuple[str, str], source_message_id: str) -> None:
if self._main_loop and self._main_loop.is_running():
fut = asyncio.run_coroutine_threadsafe(self._flush_pending_inbound_batch_after(key, source_message_id), self._main_loop)
fut.add_done_callback(lambda f, mid=source_message_id: self._log_future_error(f, "flush_inbound_batch", mid))
else:
logger.warning("[Feishu] main loop not running, cannot flush inbound batch")
def _queue_file_inbound_batch(self, msg_id: str, inbound: InboundMessage) -> bool:
key = self._pending_key(inbound.chat_id, inbound.user_id)
should_schedule_flush = False
expired_batch: tuple[str, InboundMessage, list[str]] | None = None
with self._thread_lock:
batch = self._pending_inbound_batches.get(key)
now = time.time()
if batch:
if now - batch["created_at"] <= FEISHU_INBOUND_BATCH_WINDOW_SECONDS:
batched_inbound = batch["inbound"]
batch["message_ids"].append(msg_id)
batch["text_parts"].append(inbound.text)
batched_inbound.text = "\n\n".join(part for part in batch["text_parts"] if part)
batched_inbound.files.extend(inbound.files)
batched_inbound.metadata["batched_message_ids"] = list(batch["message_ids"])
logger.info(
"[Feishu] batched inbound file message: chat_id=%s user_id=%s anchor=%s msg_id=%s files=%d",
inbound.chat_id,
inbound.user_id,
batch["anchor_message_id"],
msg_id,
len(batched_inbound.files),
)
return True
expired_batch = (batch["anchor_message_id"], batch["inbound"], list(batch["message_ids"]))
self._pending_inbound_batches[key] = {
"anchor_message_id": msg_id,
"created_at": now,
"inbound": inbound,
"message_ids": [msg_id],
"text_parts": [inbound.text],
}
inbound.metadata["batched_message_ids"] = [msg_id]
should_schedule_flush = True
if should_schedule_flush:
self._schedule_batch_flush(key, msg_id)
if expired_batch:
anchor_message_id, expired_inbound, source_message_ids = expired_batch
self._schedule_prepare_inbound(anchor_message_id, expired_inbound, source_message_ids=source_message_ids)
return True
def _pop_pending_inbound_batch(self, key: tuple[str, str], *, anchor_message_id: str | None = None) -> tuple[str, InboundMessage, list[str]] | None:
with self._thread_lock:
batch = self._pending_inbound_batches.get(key)
if not batch:
return None
if anchor_message_id is not None and batch["anchor_message_id"] != anchor_message_id:
return None
self._pending_inbound_batches.pop(key, None)
return batch["anchor_message_id"], batch["inbound"], list(batch["message_ids"])
async def _flush_pending_inbound_batch_after(self, key: tuple[str, str], anchor_message_id: str) -> None:
await asyncio.sleep(FEISHU_INBOUND_BATCH_WINDOW_SECONDS)
batch = self._pop_pending_inbound_batch(key, anchor_message_id=anchor_message_id)
if not batch:
return
anchor_message_id, inbound, source_message_ids = batch
logger.info(
"[Feishu] flushing inbound file batch: chat_id=%s user_id=%s anchor=%s messages=%d files=%d",
inbound.chat_id,
inbound.user_id,
anchor_message_id,
len(source_message_ids),
len(inbound.files),
)
await self._prepare_inbound(anchor_message_id, inbound, source_message_ids=source_message_ids)
@staticmethod
def _log_task_error(task: asyncio.Task, name: str, msg_id: str) -> None:
"""Callback for background asyncio tasks to surface errors."""
@ -727,11 +836,13 @@ class FeishuChannel(Channel):
except Exception:
pass
async def _prepare_inbound(self, msg_id: str, inbound) -> None:
async def _prepare_inbound(self, msg_id: str, inbound, *, source_message_ids: list[str] | None = None) -> None:
"""Kick off Feishu side effects without delaying inbound dispatch."""
inbound = await self._attach_connection_identity(inbound)
reaction_task = asyncio.create_task(self._add_reaction(msg_id, "OK"))
self._track_background_task(reaction_task, name="add_reaction", msg_id=msg_id)
reaction_message_ids = source_message_ids or [msg_id]
for reaction_message_id in reaction_message_ids:
reaction_task = asyncio.create_task(self._add_reaction(reaction_message_id, "OK"))
self._track_background_task(reaction_task, name="add_reaction", msg_id=reaction_message_id)
self._ensure_running_card_started(msg_id)
await self.bus.publish_inbound(inbound)
@ -922,12 +1033,17 @@ class FeishuChannel(Channel):
)
inbound.topic_id = topic_id
# Schedule on the async event loop
if self._main_loop and self._main_loop.is_running():
logger.info("[Feishu] publishing inbound message to bus (type=%s, msg_id=%s)", msg_type.value, msg_id)
fut = asyncio.run_coroutine_threadsafe(self._prepare_inbound(msg_id, inbound), self._main_loop)
fut.add_done_callback(lambda f, mid=msg_id: self._log_future_error(f, "prepare_inbound", mid))
else:
logger.warning("[Feishu] main loop not running, cannot publish inbound message")
if self._is_batchable_file_inbound(
msg_type=msg_type,
text=text,
files=files_list,
root_id=root_id,
parent_id=parent_id,
thread_id=feishu_thread_id,
):
self._queue_file_inbound_batch(msg_id, inbound)
return
self._schedule_prepare_inbound(msg_id, inbound)
except Exception:
logger.exception("[Feishu] error processing message")

171
backend/tests/test_feishu_parser.py
View file

@ -228,6 +228,7 @@ def test_feishu_on_message_extracts_image_and_file_keys():
assert files == [{"image_key": "img_123"}, {"file_key": "file_456"}]
assert "[image]" in mock_make_inbound.call_args[1]["text"]
assert "[file]" in mock_make_inbound.call_args[1]["text"]
assert channel._pending_inbound_batches == {}
def test_feishu_on_message_reuses_stored_parent_topic_for_card_replies():
@ -285,6 +286,176 @@ def _make_text_event(
return event
def _make_file_event(
file_key: str,
*,
chat_id: str = "chat_1",
message_id: str = "msg_1",
user_id: str = "user_1",
root_id: str | None = None,
parent_id: str | None = None,
thread_id: str | None = None,
):
event = MagicMock()
event.event.message.chat_id = chat_id
event.event.message.message_id = message_id
event.event.message.root_id = root_id
event.event.message.parent_id = parent_id
event.event.message.thread_id = thread_id
event.event.sender.sender_id.open_id = user_id
event.event.message.content = json.dumps({"file_key": file_key})
return event
def test_feishu_batches_top_level_file_messages_from_same_user(monkeypatch):
async def go():
monkeypatch.setattr("app.channels.feishu.FEISHU_INBOUND_BATCH_WINDOW_SECONDS", 0.01)
bus = MessageBus()
channel = FeishuChannel(bus, {"app_id": "test", "app_secret": "test"})
channel._main_loop = asyncio.get_running_loop()
channel._add_reaction = AsyncMock()
channel._ensure_running_card_started = MagicMock()
channel._on_message(_make_file_event("file_a", message_id="msg_file_1"))
channel._on_message(_make_file_event("file_b", message_id="msg_file_2"))
inbound = await asyncio.wait_for(bus.get_inbound(), timeout=0.5)
assert inbound.thread_ts == "msg_file_1"
assert inbound.topic_id == "msg_file_1"
assert inbound.text == "[file]\n\n[file]"
assert inbound.files == [{"file_key": "file_a"}, {"file_key": "file_b"}]
assert inbound.metadata["message_id"] == "msg_file_1"
assert inbound.metadata["topic_id"] == "msg_file_1"
assert inbound.metadata["batched_message_ids"] == ["msg_file_1", "msg_file_2"]
channel._ensure_running_card_started.assert_called_once_with("msg_file_1")
assert [call.args for call in channel._add_reaction.call_args_list] == [
("msg_file_1", "OK"),
("msg_file_2", "OK"),
]
_run(go())
def test_feishu_rich_text_file_message_does_not_enter_batch():
bus = MessageBus()
channel = FeishuChannel(bus, {"app_id": "test", "app_secret": "test"})
channel._schedule_prepare_inbound = MagicMock()
event = MagicMock()
event.event.message.chat_id = "chat_1"
event.event.message.message_id = "msg_rich_file"
event.event.message.root_id = None
event.event.message.parent_id = None
event.event.message.thread_id = None
event.event.sender.sender_id.open_id = "user_1"
event.event.message.content = json.dumps(
{
"content": [
[
{"tag": "text", "text": "Review"},
{"tag": "file", "file_key": "file_a"},
]
]
}
)
channel._on_message(event)
channel._schedule_prepare_inbound.assert_called_once()
inbound = channel._schedule_prepare_inbound.call_args.args[1]
assert inbound.text == "Review [file]"
assert inbound.files == [{"file_key": "file_a"}]
assert channel._pending_inbound_batches == {}
def test_feishu_file_batch_window_expiry_starts_new_topic(monkeypatch):
async def go():
monkeypatch.setattr("app.channels.feishu.FEISHU_INBOUND_BATCH_WINDOW_SECONDS", 0.01)
bus = MessageBus()
channel = FeishuChannel(bus, {"app_id": "test", "app_secret": "test"})
channel._main_loop = asyncio.get_running_loop()
channel._add_reaction = AsyncMock()
channel._ensure_running_card_started = MagicMock()
channel._on_message(_make_file_event("file_a", message_id="msg_file_1"))
first = await asyncio.wait_for(bus.get_inbound(), timeout=0.5)
channel._on_message(_make_file_event("file_b", message_id="msg_file_2"))
second = await asyncio.wait_for(bus.get_inbound(), timeout=0.5)
assert first.topic_id == "msg_file_1"
assert first.files == [{"file_key": "file_a"}]
assert second.topic_id == "msg_file_2"
assert second.files == [{"file_key": "file_b"}]
assert channel._ensure_running_card_started.call_args_list[0].args == ("msg_file_1",)
assert channel._ensure_running_card_started.call_args_list[1].args == ("msg_file_2",)
_run(go())
def test_feishu_explicit_file_reply_does_not_enter_batch(monkeypatch):
async def go():
monkeypatch.setattr("app.channels.feishu.FEISHU_INBOUND_BATCH_WINDOW_SECONDS", 10.0)
bus = MessageBus()
channel = FeishuChannel(bus, {"app_id": "test", "app_secret": "test"})
channel._main_loop = asyncio.get_running_loop()
channel._add_reaction = AsyncMock()
channel._ensure_running_card_started = MagicMock()
channel._on_message(_make_file_event("file_a", message_id="msg_reply", root_id="msg_root"))
inbound = await asyncio.wait_for(bus.get_inbound(), timeout=0.5)
assert inbound.thread_ts == "msg_reply"
assert inbound.topic_id == "msg_root"
assert inbound.files == [{"file_key": "file_a"}]
assert channel._pending_inbound_batches == {}
_run(go())
def test_feishu_expired_file_batch_does_not_get_overwritten(monkeypatch):
monkeypatch.setattr("app.channels.feishu.FEISHU_INBOUND_BATCH_WINDOW_SECONDS", 0.5)
now = 0.0
monkeypatch.setattr("app.channels.feishu.time.time", lambda: now)
bus = MessageBus()
channel = FeishuChannel(bus, {"app_id": "test", "app_secret": "test"})
channel._schedule_batch_flush = MagicMock()
channel._schedule_prepare_inbound = MagicMock()
first = InboundMessage(
channel_name="feishu",
chat_id="chat_1",
user_id="user_1",
text="[file]",
thread_ts="msg_file_1",
files=[{"file_key": "file_a"}],
)
second = InboundMessage(
channel_name="feishu",
chat_id="chat_1",
user_id="user_1",
text="[file]",
thread_ts="msg_file_2",
files=[{"file_key": "file_b"}],
)
channel._queue_file_inbound_batch("msg_file_1", first)
now = 1.0
channel._queue_file_inbound_batch("msg_file_2", second)
channel._schedule_prepare_inbound.assert_called_once_with(
"msg_file_1",
first,
source_message_ids=["msg_file_1"],
)
assert channel._pop_pending_inbound_batch(channel._pending_key("chat_1", "user_1"), anchor_message_id="msg_file_1") is None
current = channel._pop_pending_inbound_batch(channel._pending_key("chat_1", "user_1"), anchor_message_id="msg_file_2")
assert current == ("msg_file_2", second, ["msg_file_2"])
def test_feishu_plain_reply_consumes_pending_clarification_topic():
bus = MessageBus()
store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json")