{"record":{"id":"1930d6ecbc314e8e","repo":"apache/beam","slug":"langchain-is-required-to-use-langchainchunkerplease-install","errorCode":null,"errorMessage":"langchain is required to use LangChainChunkerPlease install it with using `pip install langchain`.","messagePattern":"langchain is required to use LangChainChunkerPlease install it with using `pip install langchain`\\.","errorType":"exception","errorClass":"ImportError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/ml/rag/chunking/langchain.py","lineNumber":81,"sourceCode":"      \n      with beam.Pipeline() as p:\n        chunks = (\n            p \n            | beam.Create([{'text': 'long document...', 'source': 'doc.txt'}])\n            | MLTransform(...).with_transform(chunker))\n      ```\n\n    Args:\n      text_splitter: A LangChain TextSplitter instance that defines how\n        documents are split into chunks.\n      metadata_fields: List of field names to copy from input documents to\n        chunk metadata. These fields will be preserved in each chunk created\n        from the document.\n      chunk_id_fn: Optional function that take a Chunk and return str to\n        generate chunk IDs. If not provided, random UUIDs will be used.\n    \"\"\"\n    if not TextSplitter:\n      raise ImportError(\n          \"langchain is required to use LangChainChunker\"\n          \"Please install it with using `pip install langchain`.\")\n    if not isinstance(text_splitter, TextSplitter):\n      raise TypeError(\"text_splitter must be a LangChain TextSplitter\")\n    if not document_field:\n      raise ValueError(\"document_field cannot be empty\")\n    super().__init__(chunk_id_fn)\n    self.text_splitter = text_splitter\n    self.document_field = document_field\n    self.metadata_fields = metadata_fields\n\n  def get_splitter_transform(\n      self\n  ) -> beam.PTransform[beam.PCollection[dict[str, Any]],\n                       beam.PCollection[Chunk]]:\n    return \"Langchain text split\" >> beam.ParDo(\n        _LangChainTextSplitter(\n            text_splitter=self.text_splitter,","sourceCodeStart":63,"sourceCodeEnd":99,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/ml/rag/chunking/langchain.py#L63-L99","documentation":"LangChainChunker.__init__ raises this ImportError when the langchain package is not importable in the environment. The module imports TextSplitter inside a try/except, leaving it as None on failure, and the constructor checks `if not TextSplitter` to fail fast with an actionable install hint rather than a confusing NameError later. LangChainChunker is an optional feature of the Beam RAG chunking library that delegates splitting to LangChain TextSplitters.","triggerScenarios":"Constructing LangChainChunker(text_splitter=..., document_field=...) in a Python environment where `pip install langchain` was never run, or in a deployment container/Docker image that does not include the langchain extra (apache_beam[gcp] alone does not pull it in).","commonSituations":"Deploying a Beam pipeline to Dataflow where extra packages were not passed via setup_file/requirements_file; running in a fresh venv or CI job missing the optional dependency;Beam installed without the ml/rag extras.","solutions":["Install the dependency: pip install langchain.","Add langchain to your requirements.txt / setup.py and pass it to Dataflow via --requirements_file so workers also get it.","If you don't need LangChain splitters, use a built-in chunker that has no langchain dependency."],"exampleFix":"// before\nchunker = LangChainChunker(text_splitter=RecursiveCharacterTextSplitter(...), document_field='contents')\n\n// after\n# terminal: pip install langchain\nchunker = LangChainChunker(text_splitter=RecursiveCharacterTextSplitter(...), document_field='contents')","handlingStrategy":"try-catch","validationCode":"import importlib.util\n\ndef langchain_available() -> bool:\n    return importlib.util.find_spec('langchain') is not None","typeGuard":"def is_langchain_text_splitter(obj) -> bool:\n    try:\n        from langchain_text_splitters import TextSplitter\n    except ImportError:\n        return False\n    return isinstance(obj, TextSplitter)","tryCatchPattern":"try:\n    chunker = LangChainChunker(text_splitter=splitter, document_field='contents')\nexcept ImportError:\n    logging.error('Install langchain: pip install langchain')\n    raise","preventionTips":["Pin langchain in requirements.txt and pass --requirements_file to Dataflow.","Check optional deps at pipeline-construction time, not in worker code.","Use pip check in CI to catch broken optional dependency installs."],"tags":["python","apache-beam","importerror","langchain","optional-dependency"],"backgroundTag":"missing-optional-dependency","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}