test_openai_streaming_callback() — langchain Function Reference

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

Function python LangChainCore LanguageModelBase

Entity Profile

LangChainCore→ LanguageModelBase→ test_openai_streaming_callback() — langchain Function Reference

Dependency Diagram

graph TD
  03eece37_1d9f_a7d7_58dd_65ed3c3f8feb["test_openai_streaming_callback()"]
  2edc9f83_2189_a9c8_70f5_10e2e98e8272["test_base.py"]
  03eece37_1d9f_a7d7_58dd_65ed3c3f8feb -->|defined in| 2edc9f83_2189_a9c8_70f5_10e2e98e8272
  style 03eece37_1d9f_a7d7_58dd_65ed3c3f8feb fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/partners/openai/tests/integration_tests/llms/test_base.py lines 207–221

def test_openai_streaming_callback() -> None:
    """Test that streaming correctly invokes on_llm_new_token callback."""
    callback_handler = FakeCallbackHandler()
    callback_manager = CallbackManager([callback_handler])
    llm = OpenAI(
        max_tokens=10,
        streaming=True,
        temperature=0,
        callbacks=callback_manager,
        verbose=True,
    )
    llm.invoke("Write me a sentence with 100 words.")

    # new client sometimes passes 2 tokens at once
    assert callback_handler.llm_streams >= 5