test_openai_astream() — langchain Function Reference

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

Function python LangChainCore LanguageModelBase

Entity Profile

LangChainCore→ LanguageModelBase→ test_openai_astream() — langchain Function Reference

Dependency Diagram

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

Relationship Graph

Source Code

libs/partners/openai/tests/unit_tests/chat_models/test_base.py lines 545–567

async def test_openai_astream(mock_openai_completion: list) -> None:
    llm_name = "gpt-4o"
    llm = ChatOpenAI(model=llm_name)
    assert llm.stream_usage
    mock_client = AsyncMock()

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

    mock_client.create = mock_create
    usage_chunk = mock_openai_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"]