langserve.py

Source: sunholo/streaming/langserve.py

Functions

parse_langserve_token(token)

Parses the token to accumulate content from JSON for 'event: data' events, handling JSON strings split between two tokens.

Args: token (str or bytes): The token to parse.

Returns: Generator of str: A generator that yields accumulated contents from 'event: data'.

parse_langserve_token_async(token)

Asynchronously parses the token to accumulate content from JSON for 'event: data' events, handling JSON strings split between two tokens.

Args: token (str or bytes): The token to parse.

Yields: str: An accumulated content from 'event: data'.

accumulate_json_lines(lines, start_index, run_id)

Accumulate and parse JSON string parts from a list of lines starting at a given index, using a run_id-based buffer to handle splits across tokens.

:param lines: The list of lines containing the JSON data. :param start_index: The index to start accumulation from. :param run_id: The run_id for the current JSON accumulation. :return: The accumulated JSON string if a complete JSON object is formed, or None if accumulation should continue.

parse_json_data(json_data: dict)

Attempt to parse a JSON string and yield the appropriate content or error.

:param json_data: The dict string to parse.

yields: str if within content key dict if no content key str if error in decoding json

process_langserve_lines(lines, run_id)

Process lines from langserve, parsing JSON data as needed.

:param lines: The list of lines to process. :param run_id: The current run_id to index the accumulation buffer.

process_langserve_lines_async(lines, run_id)

Asynchronously process lines from langserve, parsing JSON data as needed. This is an async wrapper for process_langserve_lines to fit into async processing.

:param lines: The list of lines to process. :param run_id: The current run_id to index the accumulation buffer.

set_metadata_value(lines)

No docstring available.