format test lint passed

This commit is contained in:
xzq.xu 2025-03-27 13:44:59 +08:00
parent d0a9808148
commit 92dc3f7341

View File

@ -17,7 +17,7 @@ from typing import (
Optional, Optional,
TypeVar, TypeVar,
Union, Union,
cast, Coroutine, cast,
) )
import pytest import pytest
@ -35,7 +35,7 @@ from langchain_core.callbacks.manager import (
CallbackManagerForRetrieverRun, CallbackManagerForRetrieverRun,
) )
from langchain_core.documents import Document from langchain_core.documents import Document
from langchain_core.messages import ToolCall, ToolMessage, AIMessage from langchain_core.messages import AIMessage, ToolCall, ToolMessage
from langchain_core.messages.tool import ToolOutputMixin from langchain_core.messages.tool import ToolOutputMixin
from langchain_core.retrievers import BaseRetriever from langchain_core.retrievers import BaseRetriever
from langchain_core.runnables import ( from langchain_core.runnables import (
@ -2609,10 +2609,15 @@ def test_title_property_preserved() -> None:
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_ainvoke_input_isolation_from_state_graph_context() -> None: async def test_ainvoke_input_isolation_from_state_graph_context() -> None:
"""Test that tool execution works correctly with LangGraph.""" """Test that tool execution works correctly with LangGraph."""
from langchain_core.messages import SystemMessage, HumanMessage, ToolMessage, AIMessage
import json import json
from blockbuster.blockbuster import blockbuster_skip from blockbuster.blockbuster import blockbuster_skip
from langchain_core.messages import (
HumanMessage,
SystemMessage,
)
blockbuster_skip.set(True) blockbuster_skip.set(True)
# Setup # Setup
@ -2634,29 +2639,29 @@ async def test_ainvoke_input_isolation_from_state_graph_context() -> None:
}, },
}, },
coroutine=sleep, coroutine=sleep,
func=sleep func=sleep,
) )
] ]
_tools = {t.name: t for t in my_tool} _tools = {t.name: t for t in my_tool}
tool_calls = ''' tool_calls = """
[{ [{
"name": "sleep", "name": "sleep",
"args": {"seconds": 2}, "args": {"seconds": 2},
"id": "call_0_82c17db8-95df-452f-a4c2-03f809022134", "id": "call_0_82c17db8-95df-452f-a4c2-03f809022134",
"type": "tool_call"}] "type": "tool_call"}]
''' """
# Test execution # Test execution
messages = [] messages: list[Any] = []
_input = "sleep for 2 seconds!" _input = "sleep for 2 seconds!"
messages.append(SystemMessage(content=prompt)) messages.append(SystemMessage(content=prompt))
messages.append(HumanMessage(content=_input)) messages.append(HumanMessage(content=_input))
ai_message = AIMessage(tool_calls=json.loads(tool_calls), content='') ai_message = AIMessage(tool_calls=json.loads(tool_calls), content="")
messages.append(ai_message) messages.append(ai_message)
result = await _tools["sleep"].ainvoke(messages[-1].tool_calls[0]["args"]) result = await _tools["sleep"].ainvoke(ai_message.tool_calls[0]["args"])
# Assertions # Assertions
assert "good" == result assert result == "good"
assert "run_manager" not in messages[-1].tool_calls[0]["args"] assert "run_manager" not in ai_message.tool_calls[0]["args"]