test_compaction_streaming() — langchain Function Reference

Architecture documentation for the test_compaction_streaming() function in test_chat_models.py from the langchain codebase.

Function python LangChainCore LanguageModelBase

Entity Profile

LangChainCore→ LanguageModelBase→ test_compaction_streaming() — langchain Function Reference

Dependency Diagram

graph TD
  63d3b2c7_4ca3_9db3_d847_fbec8f0f8f18["test_compaction_streaming()"]
  f27640dd_3870_5548_d153_f9504ae1021f["test_chat_models.py"]
  63d3b2c7_4ca3_9db3_d847_fbec8f0f8f18 -->|defined in| f27640dd_3870_5548_d153_f9504ae1021f
  style 63d3b2c7_4ca3_9db3_d847_fbec8f0f8f18 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/partners/anthropic/tests/integration_tests/test_chat_models.py lines 2467–2518

def test_compaction_streaming() -> None:
    """Test the compation beta feature."""
    llm = ChatAnthropic(
        model="claude-opus-4-6",  # type: ignore[call-arg]
        betas=["compact-2026-01-12"],
        max_tokens=4096,
        context_management={
            "edits": [
                {
                    "type": "compact_20260112",
                    "trigger": {"type": "input_tokens", "value": 50000},
                    "pause_after_compaction": False,
                }
            ]
        },
        streaming=True,
    )

    input_message = {
        "role": "user",
        "content": f"Generate a one-sentence summary of this:\n\n{'a' * 100000}",
    }
    messages: list = [input_message]

    first_response = llm.invoke(messages)
    messages.append(first_response)

    second_message = {
        "role": "user",
        "content": f"Generate a one-sentence summary of this:\n\n{'b' * 100000}",
    }
    messages.append(second_message)

    second_response = llm.invoke(messages)
    messages.append(second_response)

    content_blocks = second_response.content_blocks
    compaction_block = next(
        (block for block in content_blocks if block["type"] == "non_standard"),
        None,
    )
    assert compaction_block
    assert compaction_block["value"].get("type") == "compaction"

    third_message = {
        "role": "user",
        "content": "What are we talking about?",
    }
    messages.append(third_message)
    third_response = llm.invoke(messages)
    content_blocks = third_response.content_blocks
    assert [block["type"] for block in content_blocks] == ["text"]