langgraph/concepts/low_level/

[giscus[bot]]

langgraph/concepts/low_level/

Build language agents as graphs

https://langchain-ai.github.io/langgraph/concepts/low_level/

[pradeepjung45]

provide more code tutorials about the real world use case scenerios.

[hwchase17]

those will be in the tutorials and how-to sections, we'll add links

[siddicky]

Super excited to see Send capability! As always, Langchain Team is miles ahead of the competition.

[ArturNiklewicz]

Hi Harrison,
I just wanted to sincerely thank you for all the work you have done and shared with all of us. What you are doing is absolutely great and I wish you all the best :)

My biased & subjective junior feedback:
For most people, this should be the starting point of trying to understand how any of the code actually works.
I have just finished the LangGraph miniseries from deeplearning.ai platform, however, I had not known the abstract LangGraph concepts. The vidoes are too long for not trying to understand what is going on under the hood and too short for catching up, when so many new concepts are introduced. It is actually an intuitive practice of professionals and experienced people to explain elaborated concepts with ease in a relatively fast pace format, due to the psychological phenomena of "Curse of expertise", or "WYSIATI" (Kahneman, D. (2011). Thinking, fast and slow. Farrar, Straus and Giroux.)

I think that the most streamlined way of inviting people to experiment with LangGraph would be:

1. Showcase what can be achieved with LangGraph in a short video - a reel
2. This documentation page paired with a folder of: Jupyter notebookes of real-life popular scenarios such as the [Essay Writer, Research Assistants, etc.]

[serafinski]

Agree!

[vaibhavp4]

Couldn't find the documentation on exactly how the state is updated by the nodes.

[ArturNiklewicz]

The agent follows along the path of the compiled graph, as defined with the nodes and edges. The edges decide the direction of the next step, whereas the nodes decide the triggered logic. Each node function contains the state (TypedDict or Pydantic BaseModel) as a parameter and each node function return (a) key-value pair(s) of the state (TypedDict or Pydantic BaseModel)

[gabayben]

Please please document the low level components, Pregel, PregelNode, ChannelRead, ChannelWrite, and most importantly, channels. This is relevant for those wishing to implement their own Graph/CompiledGraph out out Pregel.

[gabayben]

out of Pregel*

[hinthornw]

Pregel isn't really a public interface and we exclude them from the docs on purpose

[gabayben]

Yeah I know but Pregel, despite being low level, is the beauty of Langgraph. StateGraph is nothing but a builder, Pregel only knows about nodes and channels. Someone can even implement a graph that resembles Haystack's Pipeline, with field-level relationships, with a compile method that returns an instance of Pregel. Which is why it should be documented.

[gabayben]

I'm just saying, you're selling this framework short. It's worth a lot more than just the StateGraph builder.

[JianXiao2021]

Hello! I would like to ask if there is a way to directly manipulate the state without invoking the reducer? I have a list like the bar in the example of the "Update state" section. I need to append element to it in a single run of the graph, but I also need to completely clear this list after the run is completed.

[donatoaz]

What is the recommended way to check if the execution was interrupted due to an "interrupt_before"? Currently I just get as a result the state as it is mid-execution of the graph. Should I have control variables in the state indicating that I hit a breakpoint?

[donatoaz]

I found in an example:

snapshot = app.get_state(thread)

if snapshot.next:
  # There was an interruption, do work ...

[Fasttyper]

I was struggling to understand the low level concepts for a while. Great doc, this was exactly what I needed. Keep up the good work!

[davlu93]

Hello,
thanks for the amazin work. I would like though to see some practical implementation for production cases where the persistence is store in a database stored in the cloud, example google big query or similar.

[Me-Baran]

Hello,

Thank you for providing this comprehensive documentation. I'm reading through the section about checkpointers and threads, and I've come across a sentence that I'm finding a bit unclear. The sentence is:

"You can use checkpointers to create threads and save the state ..."

I'm not sure I fully understand the relationship between checkpointers and threads as described here. Could you please clarify this?

Thank you for your time and for helping me understand this concept better.

[pdhoolia]

I was exploring guard-rails. Guardrails example is a great resource. The supported pattern however requires an explicit (imperative) placement of guard node in the graph. While this is sufficient for a lot of cases, I wanted to check if a more declarative placement is possible as well.

My guardrails case is that of a PII (Personally Identifiable Information). So I create a pii_guard node which wants to mask PIIs (in user_message, or tool_call results) before any of the edges target an LLM nodes. I was looking for an interrupt like declaration. Something like:

workflow.compile(
    checkpointer=memory,
    apply_before=(["agent1", "agent2"], pii_guard),
    apply_after=(["agent1", "agent2"], pii_unguard),
)

Such an aspect-oriented binding, makes it much simpler to ensure that no matter how I land up on my LLM node, the desired masking would happen, and post LLM the desired un-masking would happen. Explicit specification in the imperative flow is a bit of a pain.

[pdhoolia]

Ignore previous comment. The declarative apply_before and apply_after should already be do-able using decorators. E.g. in the python version I should already be able to do something like:

import functools

def pre(state, config):
    # Pre-processing logic
    print("Running pre-processing...")
    state['pre_processed'] = True
    return state, config

def post(state, config):
    # Post-processing logic
    print("Running post-processing...")
    state['post_processed'] = True
    return state, config

# Define a decorator that wraps pre and post logic around any function
def pre_post_decorator(func):
    @functools.wraps(func)
    def wrapper(state, config):
        # Run pre-processing logic
        state, config = pre(state, config)
        
        # Execute the original function
        result = func(state, config)
        
        # Run post-processing logic
        state, config = post(state, config)
        
        return result
    return wrapper

# Example agent functions
@pre_post_decorator
def call_agent_1(state, config):
    print(f"Executing agent 1: {state}")
    # Agent computation here
    return state

@pre_post_decorator
def call_agent_2(state, config):
    print(f"Executing agent 2: {state}")
    # Agent computation here
    return state

# Let's see this in action
state = {'pre_processed': False, 'post_processed': False}
config = {}

# These function calls will automatically run the pre and post logic
call_agent_1(state, config)
print(f"{state}")
state = {'pre_processed': False, 'post_processed': False}
call_agent_2(state, config)
print(f"{state}")

And in the javascript version, I should be able to do:

// Define the pre-processing function
function pre(state, config) {
    console.log("Running pre-processing...");
    state.preProcessed = true;
    return [state, config];
  }
  
  // Define the post-processing function
  function post(state, config) {
    console.log("Running post-processing...");
    state.postProcessed = true;
    return [state, config];
  }
  
  // Create a higher-order function (decorator) to wrap around the agent functions
  function prePostWrapper(func) {
    return function (state, config) {
      // Run pre-processing
      [state, config] = pre(state, config);
      
      // Execute the original function
      const result = func(state, config);
      
      // Run post-processing
      [state, config] = post(state, config);
      
      return result;
    };
  }
  
  // Example agent functions
  function callAgent1(state, config) {
    console.log("Executing agent 1: " + JSON.stringify(state));
    // Agent computation here
    return state;
  }
  
  function callAgent2(state, config) {
    console.log("Executing agent 2: " + JSON.stringify(state));
    // Agent computation here
    return state;
  }
  
  // Wrap the agent functions with prePostWrapper
  const wrappedAgentModel1 = prePostWrapper(callAgent1);
  const wrappedAgentModel2 = prePostWrapper(callAgent2);
  
  
  // Let's see this in action
  let state = { preProcessed: false, postProcessed: false };
  let config = {};
  
  // Use wrapped function. They will automatically run the pre and post logic
  wrappedAgentModel1(state, config);
  state = { preProcessed: false, postProcessed: false };
  wrappedAgentModel2(state, config);

[Eknathabhiram]

I want to know whether we make routing function return to different nodes for the same key .

[djprawns]

lets say i want to have a certain method to be called after every node invocation, how can i achieve that? also how can i specify that for certain nodes, lets say START and END?

[vabbybansal]

Is there any way to create dependency graph functionality as well? Currently, graph nodes do not wait for all dependencies to resolve before executing a node by design. If we can also achieve some dependency graph kind of use-cases by adding some graph param then, that would be great!

[vabbybansal]

Looks like this can be achieved by creating a multi-node edge, something like this -
graph.add_edge(["node1", "node2"], "node3")

This way, node3 waits for both node1 and node2 to finish and we can achieve task dependency properly.
Can we please add this to the main langgraph documentation? I found this on a github comment - #954

[raghuldeva]

Can someone tell how do I integrate Langgraph's Human In Loop with FastAPI. As far as I've seen there is no integration with FAST API