test_deepseek_astream() — langchain Function Reference

Architecture documentation for the test_deepseek_astream() function in test_base.py from the langchain codebase.

Function python LangChainCore LanguageModelBase

Entity Profile

LangChainCore→ LanguageModelBase→ test_deepseek_astream() — langchain Function Reference

Dependency Diagram

graph TD
  b426357d_579c_513a_8453_613becfbdcd5["test_deepseek_astream()"]
  48232d20_f8c1_b597_14fa_7dc407e9bfe5["test_base.py"]
  b426357d_579c_513a_8453_613becfbdcd5 -->|defined in| 48232d20_f8c1_b597_14fa_7dc407e9bfe5
  style b426357d_579c_513a_8453_613becfbdcd5 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/partners/openai/tests/unit_tests/chat_models/test_base.py lines 477–498

async def test_deepseek_astream(mock_deepseek_completion: list) -> None:
    llm_name = "deepseek-chat"
    llm = ChatOpenAI(model=llm_name, stream_usage=True)
    mock_client = AsyncMock()

    async def mock_create(*args: Any, **kwargs: Any) -> MockAsyncContextManager:
        return MockAsyncContextManager(mock_deepseek_completion)

    mock_client.create = mock_create
    usage_chunk = mock_deepseek_completion[-1]
    usage_metadata: UsageMetadata | None = None
    with patch.object(llm, "async_client", mock_client):
        async for chunk in llm.astream("你的名字叫什么？只回答名字"):
            assert isinstance(chunk, AIMessageChunk)
            if chunk.usage_metadata is not None:
                usage_metadata = chunk.usage_metadata

    assert usage_metadata is not None

    assert usage_metadata["input_tokens"] == usage_chunk["usage"]["prompt_tokens"]
    assert usage_metadata["output_tokens"] == usage_chunk["usage"]["completion_tokens"]
    assert usage_metadata["total_tokens"] == usage_chunk["usage"]["total_tokens"]