Streaming is not working in SagemakerEndpoint

[anserali]

Checked other resources

• I added a very descriptive title to this question.
• I searched the LangChain documentation with the integrated search.
• I used the GitHub search to find a similar question and didn't find it.

Commit to Help

• I commit to help with one of those options 👆

Example Code

from uuid import UUID
from typing import Any, Dict, List
from langchain.schema.output import LLMResult
from langchain.callbacks.base import BaseCallbackHandler
from langchain_community.llms.sagemaker_endpoint import LLMContentHandler
from langchain_community.llms.sagemaker_endpoint import SagemakerEndpoint
from langchain.prompts import PromptTemplate
from langchain.chains.question_answering import load_qa_chain
from typing import Dict
import json
from langchain.docstore.document import Document
from tenacity import RetryCallState
example_doc_1 = """
Peter and Elizabeth took a taxi to attend the night party in the city. While in the party, Elizabeth collapsed and was rushed to the hospital.
Since she was diagnosed with a brain injury, the doctor told Peter to stay besides her until she gets well.
Therefore, Peter stayed with her at the hospital for 3 days without leaving.
"""

docs = [
    Document(
        page_content=example_doc_1,
    )
]
query = """How long was Elizabeth hospitalized?
"""

prompt_template = """Use the following pieces of context to answer the question at the end.

{context}

Question: {question}
Answer:"""
PROMPT = PromptTemplate(
    template=prompt_template, input_variables=["context", "question"]
)


class MyCustomHandler(BaseCallbackHandler):
    def __init__(self) -> None:
        super().__init__()

    def on_llm_new_token(self, token: str, **kwargs) -> None:
        print("***********************")

    def on_text(self, text: str, *, run_id: UUID, parent_run_id: UUID | None = None, **kwargs: Any) -> Any:
        print("***********on text")
        return super().on_text(text, run_id=run_id, parent_run_id=parent_run_id, **kwargs)

    def on_llm_end(
        self,
        response: LLMResult,
        *,
        run_id: UUID,
        parent_run_id: UUID | None = None,
        **kwargs: Any,
    ) -> Any:
        print("************** on llm end ***********************")
        return super().on_llm_end(
            response, run_id=run_id, parent_run_id=parent_run_id, **kwargs
        )

    def on_llm_error(self, error: BaseException, *, run_id: UUID, parent_run_id: UUID | None = None, **kwargs: Any) -> Any:
        print("***************ON ERROR ********************", error)
        return super().on_llm_error(error, run_id=run_id, parent_run_id=parent_run_id, **kwargs)

    def on_retry(self, retry_state: RetryCallState, *, run_id: UUID, parent_run_id: UUID | None = None, **kwargs: Any) -> Any:
        print("***************************")
        return super().on_retry(retry_state, run_id=run_id, parent_run_id=parent_run_id, **kwargs)

    def on_llm_start(
        self,
        serialized: Dict[str, Any],
        prompts: List[str],
        *,
        run_id: UUID,
        parent_run_id: UUID | None = None,
        tags: List[str] | None = None,
        metadata: Dict[str, Any] | None = None,
        **kwargs: Any,
    ) -> Any:
        print("************on llm start ******************")
        return super().on_llm_start(
            serialized,
            prompts,
            run_id=run_id,
            parent_run_id=parent_run_id,
            tags=tags,
            metadata=metadata,
            **kwargs,
        )


class ContentHandler(LLMContentHandler):
    content_type = "application/json"
    accepts = "application/json"

    def transform_input(self, prompt: str, model_kwargs: Dict) -> bytes:
        input_str = json.dumps(
            {"inputs": prompt, **model_kwargs, "stream": True})
        return input_str.encode('utf-8')

    def transform_output(self, output: bytes) -> str:
        response_json = json.loads(output.read().decode("utf-8"))
        return response_json[0]['generated_text']


content_handler = ContentHandler()
callback_f = MyCustomHandler()

llm = SagemakerEndpoint(
    streaming=True,
    endpoint_name="endpoint",
    region_name="region",
    model_kwargs={"temperature": 0.7,
                  "max_new_tokens": 400, },
    content_handler=content_handler,
    callbacks=[callback_f],
    verbose=True
)
chain = load_qa_chain(
    llm=llm,
    prompt=PROMPT,
)

chain.invoke({"question": query, "input_documents": docs})

Description

I am trying to stream the response of sagemaker, But the code throws errors as below, seems like JSON.loads(line) from sagemaker_endpoint.py is throwing error

`Traceback (most recent call last):
File "C:\Users\NYL\AppData\Roaming\Python\Python311\site-packages\langchain_community\llms\sagemaker_endpoint.py", line 345, in call
resp = json.loads(line)
^^^^^^^^^^^^^^^^
File "C:\Users\NYL\anaconda3\envs\sagemaker\Lib\json_init.py", line 346, in loads
return _default_decoder.decode(s)
^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\NYL\anaconda3\envs\sagemaker\Lib\json\decoder.py", line 337, in decode
obj, end = self.raw_decode(s, idx=_w(s, 0).end())
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\NYL\anaconda3\envs\sagemaker\Lib\json\decoder.py", line 357, in raw_decode
raise JSONDecodeError("Expecting value", s, err.value) from None
json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "F:\poc\aws_sage_maker\sagemaker_langchain.py", line 127, in
out = chain.invoke({"question": query, "input_documents": docs})
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\NYL\AppData\Roaming\Python\Python311\site-packages\langchain\chains\base.py", line 162, in invoke
raise e
File "C:\Users\NYL\AppData\Roaming\Python\Python311\site-packages\langchain\chains\base.py", line 156, in invoke
self._call(inputs, run_manager=run_manager)
File "C:\Users\NYL\AppData\Roaming\Python\Python311\site-packages\langchain\chains\combine_documents\base.py", line 136, in _call
output, extra_return_dict = self.combine_docs(
^^^^^^^^^^^^^^^^^^
File "C:\Users\NYL\AppData\Roaming\Python\Python311\site-packages\langchain\chains\combine_documents\stuff.py", line 244, in combine_docs
return self.llm_chain.predict(callbacks=callbacks, **inputs), {}
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\NYL\AppData\Roaming\Python\Python311\site-packages\langchain\chains\llm.py", line 293, in predict
return self(kwargs, callbacks=callbacks)[self.output_key]
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\NYL\anaconda3\envs\sagemaker\Lib\site-packages\langchain_core_api\deprecation.py", line 145, in warning_emitting_wrapper
return wrapped(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\NYL\AppData\Roaming\Python\Python311\site-packages\langchain\chains\base.py", line 363, in call
return self.invoke(
^^^^^^^^^^^^
File "C:\Users\NYL\AppData\Roaming\Python\Python311\site-packages\langchain\chains\base.py", line 162, in invoke
raise e
File "C:\Users\NYL\AppData\Roaming\Python\Python311\site-packages\langchain\chains\base.py", line 156, in invoke
self._call(inputs, run_manager=run_manager)
File "C:\Users\NYL\AppData\Roaming\Python\Python311\site-packages\langchain\chains\llm.py", line 103, in _call
response = self.generate([inputs], run_manager=run_manager)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\NYL\AppData\Roaming\Python\Python311\site-packages\langchain\chains\llm.py", line 115, in generate
return self.llm.generate_prompt(
^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\NYL\anaconda3\envs\sagemaker\Lib\site-packages\langchain_core\language_models\llms.py", line 568, in generate_prompt
return self.generate(prompt_strings, stop=stop, callbacks=callbacks, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\NYL\anaconda3\envs\sagemaker\Lib\site-packages\langchain_core\language_models\llms.py", line 741, in generate
output = self._generate_helper(
^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\NYL\anaconda3\envs\sagemaker\Lib\site-packages\langchain_core\language_models\llms.py", line 605, in _generate_helper
raise e
File "C:\Users\NYL\anaconda3\envs\sagemaker\Lib\site-packages\langchain_core\language_models\llms.py", line 592, in _generate_helper
self._generate(
File "C:\Users\NYL\anaconda3\envs\sagemaker\Lib\site-packages\langchain_core\language_models\llms.py", line 1177, in _generate
self._call(prompt, stop=stop, run_manager=run_manager, **kwargs)
File "C:\Users\NYL\AppData\Roaming\Python\Python311\site-packages\langchain_community\llms\sagemaker_endpoint.py", line 355, in _call
raise ValueError(f"Error raised by streaming inference endpoint: {e}")
ValueError: Error raised by streaming inference endpoint: Expecting value: line 1 column 1 (char 0)`

System Info

Library used

aiobotocore==2.7.0
aiohttp==3.9.1
aioitertools==0.11.0
aiosignal==1.3.1
annotated-types==0.6.0
anyio==4.2.0
apify-client==1.5.0
apify-shared==1.0.4
asgiref==3.7.2
attrs==23.2.0
backoff==2.2.1
beautifulsoup4==4.12.3
blinker==1.7.0
boto3==1.34.64
botocore==1.34.64
build==1.0.3
certifi==2023.11.17
chardet==5.2.0
charset-normalizer==3.3.2
chromadb==0.4.22
chromadb-client==0.4.23.dev0
click==8.1.7
coloredlogs==15.0.1
dataclasses-json==0.6.3
distro==1.9.0
fastapi==0.109.0
filelock==3.13.1
Flask==3.0.0
frozenlist==1.4.1
google-auth==2.26.2
googleapis-common-protos==1.62.0
greenlet==3.0.3
httpcore==1.0.2
httpx==0.26.0
huggingface-hub==0.20.2
idna==3.6
importlib-metadata==6.11.0
itsdangerous==2.1.2
Jinja2==3.1.3
jmespath==1.0.1
joblib==1.3.2
jsonpatch==1.33
jsonpointer==2.4
kubernetes==29.0.0
langchain==0.1.1
langchain-community==0.0.13
langchain-core==0.1.23
langchain-experimental==0.0.49
langsmith==0.0.87
loguru==0.7.2
lxml==5.1.0
MarkupSafe==2.1.3
marshmallow==3.20.2
multidict==6.0.4
mypy-extensions==1.0.0
mysqlclient==2.2.1
numpy==1.26.3
onnxruntime==1.16.3
opentelemetry-api==1.22.0
opentelemetry-exporter-otlp-proto-common==1.22.0
opentelemetry-exporter-otlp-proto-grpc==1.22.0
opentelemetry-instrumentation==0.43b0
opentelemetry-instrumentation-asgi==0.43b0
opentelemetry-instrumentation-fastapi==0.43b0
opentelemetry-proto==1.22.0
opentelemetry-sdk==1.22.0
packaging==23.2
pandas==2.1.4
pinecone-client==2.2.4
posthog==3.3.1
protobuf==4.25.2
pyasn1==0.5.1
pyasn1-modules==0.3.0
pydantic==2.6.4
pydantic_core==2.16.3
python-dateutil==2.9.0.post0
python-docx==1.1.0
PyYAML==6.0.1
requests==2.31.0
requests-oauthlib==1.3.1
s3fs==2023.10.0
s3transfer==0.10.1
six==1.16.0
sniffio==1.3.1
SQLAlchemy==2.0.25
starlette==0.35.1
tenacity==8.2.3
tokenizers==0.15.0
typer==0.9.0
typing-inspect==0.9.0
typing_extensions==4.10.0
unstructured==0.12.0
unstructured-client==0.15.2
urllib3==2.2.1
uvicorn==0.26.0
watchfiles==0.21.0
Werkzeug==3.0.1
yarl==1.9.4

[dosubot[bot]]

Hey there @anserali! 👋 I'm Dosu, a friendly bot here to help you tackle bugs, answer your questions, and guide you on your journey to becoming a contributor while we wait for a human maintainer to join us. I'm diving into your issue and will get back to you with a full answer shortly. Stay tuned!

[anserali]

ok

[dosubot[bot]]

The error you're encountering, Expecting value: line 1 column 1 (char 0), typically occurs when you're trying to parse an empty string as JSON. In your case, this error might be happening in the streaming section where json.loads(line) is used to parse each line of the response from the Sagemaker endpoint.

There are a few possible reasons for this error:

1. Empty Line: If the LineIterator yields an empty line (which should not normally happen given the logic to only return lines ending with a newline character), attempting to parse it as JSON would result in the mentioned error.

2. Incomplete JSON Data: The LineIterator is designed to handle cases where JSON objects may be split across multiple PayloadPart events. If there's a logic error or unexpected behavior in how bytes are concatenated or lines are yielded (for example, yielding a line before it's fully complete), it could result in attempting to parse an incomplete JSON string, leading to the error.

3. Incorrectly Formatted Response: If the response from the Sagemaker endpoint is not correctly formatted (e.g., not properly JSON serialized), attempting to parse it as JSON would also result in an error. However, this seems less likely given the specific error message and the fact that the response format is likely controlled and consistent.

To diagnose and resolve this issue, consider the following steps:

• Debug Logging: Temporarily add logging before the json.loads(line) call to print out the line being parsed. This can help confirm whether the issue is due to empty lines or improperly concatenated JSON strings.

• Review Endpoint Response: Ensure that the Sagemaker endpoint is configured to return responses in the expected format, and that there are no issues with how the response is generated on the endpoint's side.

• Error Handling: Implement error handling around the json.loads(line) call to gracefully handle cases where parsing fails, potentially logging or skipping lines that cannot be parsed.

Here's an example of how you might add logging and error handling:

import logging

# Example of adding logging before parsing the line
for line in iterator:
    logging.debug(f"Attempting to parse line: {line}")
    try:
        resp = json.loads(line)
        # Rest of the code follows...
    except json.JSONDecodeError as e:
        logging.error(f"Failed to parse line as JSON: {line}. Error: {e}")
        continue  # Skip this line or handle the error as appropriate

This approach can help identify the problematic line(s) and guide further investigation or mitigation strategies.

For more details, you can refer to the SagemakerEndpoint class in the LangChain codebase.

Sources

• libs/community/langchain_community/llms/sagemaker_endpoint.py
• libs/community/langchain_community/embeddings/sagemaker_endpoint.py

About Dosu

This response is meant to be useful and save you time. It is not meant to be a precise solution, but rather a starting point for your own research.

Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.

If you want to continue the conversation, start your reply with @dosu-bot.

[anserali]

I printed the line it is
data:{"index":1,"token":{"id":13,"text":"\n","logprob":-2.7832031,"special":false},"generated_text":null,"details":null} 0
could you help me what is the problem with the line?
even it is an how am I suppose to change on a library ?

[mohitsaini-tally]

@anserali I'm also facing the same issue. did you find any solution for this?