langchain-ai/langserve
1
0
Fork 0
mirror of https://github.com/langchain-ai/langserve.git synced 2026-07-16 09:34:30 -04:00
Code Issues 136 Packages Projects Releases 10 Wiki Activity

Running background tasks #202

New Issue
Open
opened 2026-02-16 00:19:23 -05:00 by yindo · 2 comments
yindo commented 2026-02-16 00:19:23 -05:00
Owner
Copy Link
Copy Source

Originally created by @Sara2823 on GitHub (Jun 27, 2024).

How to add a FastAPI BackgroundTask to a LangServe endpoint?

We need to save chat history to a database after each request, and not block the user input until the messages are saved.

This is how we add the endpoint to the app:

add_routes(
    app,
    agent,
    path="/agent",
    per_req_config_modifier=per_request_config_modifier,
    config_keys=["configurable", "metadata"]
)

Thanks

Originally created by @Sara2823 on GitHub (Jun 27, 2024). How to add a FastAPI BackgroundTask to a LangServe endpoint? We need to save chat history to a database after each request, and not block the user input until the messages are saved. This is how we add the endpoint to the app: ``` add_routes( app, agent, path="/agent", per_req_config_modifier=per_request_config_modifier, config_keys=["configurable", "metadata"] ) ``` Thanks
yindo commented 2026-02-16 00:19:23 -05:00
Author
Owner
Copy Link
Copy Source

@eyurtsev commented on GitHub (Jun 27, 2024):

Hi @Sara2823 you can use the APIHandler directly if you need fine grained control over the endpoint definition (e.g., to expose background tasks).

@eyurtsev commented on GitHub (Jun 27, 2024): Hi @Sara2823 you can use the `APIHandler` directly if you need fine grained control over the endpoint definition (e.g., to expose background tasks).
yindo commented 2026-02-16 00:19:23 -05:00
Author
Owner
Copy Link
Copy Source

@Sara2823 commented on GitHub (Jul 7, 2024):

Thanks @eyurtsev
This code works but I had to copy the whole streaming function to the route to put the insertion function(insert_message) in the middle of the process. Is there a better solution I'm missing?

api_handler = APIHandler(config_agent,path ="/agent", per_req_config_modifier=per_request_config_modifier)
@app.post("/agent/stream_log")
async def stream_endpoint(request: Request, background_tasks: BackgroundTasks):
    # Get user input and configuration
    try:
        config, input_ = await api_handler._get_config_and_input(
            request,
            "",
            endpoint="stream_log",
            server_config=None,
        )
    except BaseException:
        raise
    try:
        body = await request.json()
        with _with_validation_error_translation():
            stream_log_request = StreamLogParameters(**body)
    except json.JSONDecodeError:
        raise RequestValidationError(errors=["Invalid JSON body"])
    except RequestValidationError:
        raise

            # Streaming loop
    async def _stream():
        try:
            async for chunk in api_handler._runnable.astream_log(
                input_,
                config=config,
                diff=True,
                include_names=stream_log_request.include_names,
                include_types=stream_log_request.include_types,
                include_tags=stream_log_request.include_tags,
                exclude_names=stream_log_request.exclude_names,
                exclude_types=stream_log_request.exclude_types,
                exclude_tags=stream_log_request.exclude_tags,
            ):
                if not isinstance(chunk, RunLogPatch):
                    raise AssertionError(
                        f"Expected a RunLog instance got {type(chunk)}"
                    )
                if (
                    api_handler._names_in_stream_allow_list is None
                    or api_handler._runnable.config.get("run_name")
                    in api_handler._names_in_stream_allow_list
                ):
                    data = {
                        "ops": chunk.ops,
                    }
                    yield {

                        "data": api_handler._serializer.dumps(data).decode("utf-8"),
                        "event": "data",
                    }
            final_answer = chunk.ops[0]['value']['output'].return_values['output']
            background_tasks.add_task(insert_message, config['configurable']['user_id'],"conversational",  input_['input'], final_answer)
            
            
        except BaseException:
            yield {
                "event": "error",
                "data": json.dumps(
                    {"status_code": 500, "message": "Internal Server Error"}
                ),
            }
            raise
    return EventSourceResponse(_stream())
@Sara2823 commented on GitHub (Jul 7, 2024): Thanks @eyurtsev This code works but I had to copy the whole streaming function to the route to put the insertion function(`insert_message`) in the middle of the process. Is there a better solution I'm missing? ``` api_handler = APIHandler(config_agent,path ="/agent", per_req_config_modifier=per_request_config_modifier) @app.post("/agent/stream_log") async def stream_endpoint(request: Request, background_tasks: BackgroundTasks): # Get user input and configuration try: config, input_ = await api_handler._get_config_and_input( request, "", endpoint="stream_log", server_config=None, ) except BaseException: raise try: body = await request.json() with _with_validation_error_translation(): stream_log_request = StreamLogParameters(**body) except json.JSONDecodeError: raise RequestValidationError(errors=["Invalid JSON body"]) except RequestValidationError: raise # Streaming loop async def _stream(): try: async for chunk in api_handler._runnable.astream_log( input_, config=config, diff=True, include_names=stream_log_request.include_names, include_types=stream_log_request.include_types, include_tags=stream_log_request.include_tags, exclude_names=stream_log_request.exclude_names, exclude_types=stream_log_request.exclude_types, exclude_tags=stream_log_request.exclude_tags, ): if not isinstance(chunk, RunLogPatch): raise AssertionError( f"Expected a RunLog instance got {type(chunk)}" ) if ( api_handler._names_in_stream_allow_list is None or api_handler._runnable.config.get("run_name") in api_handler._names_in_stream_allow_list ): data = { "ops": chunk.ops, } yield { "data": api_handler._serializer.dumps(data).decode("utf-8"), "event": "data", } final_answer = chunk.ops[0]['value']['output'].return_values['output'] background_tasks.add_task(insert_message, config['configurable']['user_id'],"conversational", input_['input'], final_answer) except BaseException: yield { "event": "error", "data": json.dumps( {"status_code": 500, "message": "Internal Server Error"} ), } raise return EventSourceResponse(_stream()) ```
Sign in to join this conversation.
No Branch/Tag Specified
Branches Tags
Labels
Clear labels
bug
confirmed
duplicate
enhancement
help wanted
in-progress
investigate
not langserve
playground
pull-request
Mirrored from GitHub Pull Request
 question
stale
streaming
unable-to-reproduce
wontfix
No Label
Milestone
No items
No Milestone
Projects
Clear projects
No project
Assignees
Clear assignees
yindo
No Assignees
1 Participants
Notifications
Due Date
No due date set.
Dependencies

No dependencies set.

Reference: langchain-ai/langserve#202
Delete Branch "%!s()"

Deleting a branch is permanent. Although the deleted branch may continue to exist for a short time before it actually gets removed, it CANNOT be undone in most cases. Continue?