mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-11-13 06:31:47 +00:00
3d826c6d95
* feat(hitl): add human-in-the-loop approval for high-risk tool calls * fix(hitl): address approval review findings * docs(hitl): keep agent guidance under size limit * fix(hitl): guard approval pause recovery * docs(hitl): shrink inherited guidance * fix(hitl): harden approval admission and recovery * fix(approvals): address review safety and reconciliation gaps * ci: retry approval checks after runner timeout --------- Co-authored-by: wangzeren <1004695029@qq.com>
19 lines
820 B
Python
19 lines
820 B
Python
"""Scripted model shared by execution-boundary approval tests."""
|
|
|
|
from typing import Any
|
|
|
|
from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
|
|
from langchain_core.messages import AIMessage, ToolMessage
|
|
from langchain_core.outputs import ChatGeneration, ChatResult
|
|
|
|
|
|
class StateAwareToolModel(FakeMessagesListChatModel):
|
|
calls: list[dict[str, Any]]
|
|
|
|
def bind_tools(self, tools: Any, **kwargs: Any) -> Any:
|
|
return self
|
|
|
|
def _generate(self, messages: Any, stop: Any = None, run_manager: Any = None, **kwargs: Any) -> ChatResult:
|
|
message = AIMessage(content="final answer") if any(isinstance(message, ToolMessage) for message in messages) else AIMessage(content="", tool_calls=self.calls)
|
|
return ChatResult(generations=[ChatGeneration(message=message)])
|