langgraph/cloud/quick_start/

[giscus[bot]]

langgraph/cloud/quick_start/

Build language agents as graphs

https://langchain-ai.github.io/langgraph/cloud/quick_start/

[Jaid844]

Let's goo yeahh!!

[DhavalThkkar]

So there'll be no way to deploy this using Langserve locally?
Can I get it working without LangSmith?

[hinthornw]

You can use langserve or your own fastAPI endpoint - it just won't get the additional cloud features we've built in

[DhavalThkkar]

Any samples that you'd recommend? There seems to be little to no documentation available for that and since there is state management in langgraph, integrating with langserve becomes equally challenging

[vbarda]

@DhavalThkkar we just added more deployment options for LangGraph Platform (formerly LangGraph Cloud), including "Self-hosted Lite" - a limited free version. It is available to anyone with a free (developer) LangSmith account. Please check out the documentation to learn more: https://langchain-ai.github.io/langgraph/concepts/deployment_options/#self-hosted-lite. let me know if you have any questions!

[damianoneill]

@hinthornw can you provide an example of integrating with langserve?

[rogerthatdev]

Hi!

When accessing our agent via the SDK, we're not seeing the same object returned that's in these docs.

{'agent': {'messages': [{'content': "Hi Bagatur! It's nice to meet you. How can I assist you today?", 'additional_kwargs': {}, 'response_metadata': {'finish_reason': 'stop', 'model_name': 'gpt-4o-2024-05-13', 'system_fingerprint': 'fp_9cb5d38cf7'}, 'type': 'ai', 'name': None, 'id': 'run-c89118b7-1b1e-42b9-a85d-c43fe99881cd', 'example': False, 'tool_calls': [], 'invalid_tool_calls': [], 'usage_metadata': None}]}}

by any chance, are these docs up to date? Are there additional SDK docs for Javascript?

[lvyoudashuju]

@hinthornw can you provide an example of integrating with langserve or?
or can you provide an example of own fastAPI endpoint?

[fletchertyler914]

Here's my current solution to streaming events from langgraph via a langserve project. You could simplify it a lot more, as im dealing with multiple tools in a ReAct style graph now. I originally used langserve with pure LCEL before langgraph came out, and ended up converting to langgraph ad use SSE to stream all LLM and tool tokens. Here's an overview of the API side:

# Load environment variables from .env file before importing any other modules
from dotenv import load_dotenv
load_dotenv(override=True)  # noqa

from fastapi import APIRouter, FastAPI, Request, Response
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import RedirectResponse, StreamingResponse

from api.models.api import GraphRequest


# Create a router for all /api routes
router = APIRouter()


# Redirect the root to the documentation
@router.get("/")
async def redirect_root_to_docs():
    return RedirectResponse("/api/docs")


@router.get("/health-check")
def readiness_probe():
    return Response(status_code=200, content="API is healthy")


@router.post("/v1/chat")
async def chat_stream(
    request: Request,
    body: GraphRequest
):
    try:
        # Get the server config from the request (per user)
        server_config = await per_req_config_modifier({}, request)
        # print("Chat Req Server Config:", server_config)

        return StreamingResponse(
            generate_chat_events(
                message=body.message,
                config=server_config,
            ),
            media_type="text/event-stream"
        )
    except Exception as e:
        return StreamingResponse(
            content=f"event: error\ndata: {str(e)}\n\n",
            status_code=500,
            media_type="text/event-stream"
        )

# Create the FastAPI app
app = FastAPI(
    title="API Server",
    version="1.0",
    description="API server for LangGraph",
    root_path="/api"
)

# Mount the router under the /api prefix
app.include_router(router)

# Set all CORS enabled origins
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
    expose_headers=["*"],
)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(
        app,
        host="0.0.0.0",
        port=8001,
        proxy_headers=True,
        forwarded_allow_ips="*"
    )

where my generate_chat_events is essentially a middleware to customize/filter the event stream. It looks like this:

async def generate_chat_events(message: str, config: RunnableConfig):
    # print("Generating Chat Events", message, config)
    async for event in graph.astream_events(
        input=format_user_input(message),
        config=config,
        version="v2",
    ):
        kind = event["event"]
        tags = event.get("tags", [])

        # Graph Stream
        if kind == "on_chat_model_stream" and "chat_node" in tags:
            # Check if the chunk has content
            chunk = event["data"]["chunk"]

            if chunk.content:
                # Serialize the chunk content
                chunk_content = serialize_ai_message_chunk(chunk)

                # Yield the formatted SSE content
                yield build_response_message(
                    chunk_content=chunk_content,
                    config=config
                )

       # Here I have tool calls that I want to yield events to show a 'status' update in the UI. This is optional, only if you have tools and want to know when they stop/start.

        # Start Tool Status
        elif kind == "on_tool_start":
            if event["name"] == "guest_search_tool":
                yield build_response_message(
                    chunk_content="\nFinding Matching Guest... ",
                    config=config
                )
            elif event["name"] == "guest_sample_tool":
                yield build_response_message(
                    chunk_content="\n\nGetting Guest Sample... ",
                    # reset the guest sample
                    guest_sample=[],
                    config=config
                )

        # End Tool Status
        elif kind == "on_tool_end":
            if event["name"] == "guest_search_tool":
                yield build_response_message(
                    # chunk_content="Done!\n\n",
                    chunk_content="✅\n\n",
                    config=config
                )
            elif event["name"] == "guest_sample_tool":
                # Get the guest sample from the event data
                event_data = event["data"]["output"]
                # print(event_data)
                guest_sample = event_data.artifact

                yield build_response_message(
                    # chunk_content="Done!\n",
                    chunk_content="✅\n\n",
                    guest_sample=guest_sample,
                    config=config
                )

here are the helper functions:

def serialize_ai_message_chunk(chunk):
    """
    Custom serializer for AIMessageChunk objects.
    Convert the AIMessageChunk object to a serializable format.
    """
    if isinstance(chunk, AIMessageChunk):
        return chunk.content
    else:
        raise TypeError(
            f"Object of type {type(chunk).__name__} is not correctly formatted for serialization"
        )


def format_user_input(content: str):
    return {
        "messages": [
            HumanMessage(content)
        ]
    }


def build_response_message(
        config: RunnableConfig,
        chunk_content: str = None,
        guest_sample=None,
        messages=None,
):
    # yield the current state after each event
    graph_state = graph.get_state(config).values
    messages = graph_state.get("messages", [])

    # convert the messages to a serializable format
    messages = [
        {
            "role": message.type,
            "content": message.content
        }
        for message in messages
    ]

    # Properly format the content for SSE
   # This is my custom response type, you can return whatever you want here, or only the chunk_content!
    data = json.dumps({
        "content": chunk_content,
        "guest_sample": guest_sample,
        "messages": messages,
    })
    # Format the event message
    return f"event: message\ndata: {data}\n\n"

You can just return the graph stream or stream_events, but its helpful to format/filter the stream events in the api side so your frontend only gets the events you want displayed. I may put a simple repo up to show this in action.

[think-circle]

Has anyone got it working on langserve both from python backend and react js front end ?

[vbarda]

Hi folks! We just added more deployment options for LangGraph Platform (formerly LangGraph Cloud), including "Self-hosted Lite" - a limited free version. It is available to anyone with a free (developer) LangSmith account. Please check out the documentation to learn more: https://langchain-ai.github.io/langgraph/concepts/deployment_options/#self-hosted-lite. let me know if you have any questions!