create_agent as a langgraph node

[EmmetWilliam]

Checked other resources

• This is a bug, not a usage question. For questions, please use the LangChain Forum (https://forum.langchain.com/).
• I added a clear and detailed title that summarizes the issue.
• I read what a minimal reproducible example is (https://stackoverflow.com/help/minimal-reproducible-example).
• I included a self-contained, minimal example that demonstrates the issue INCLUDING all the relevant imports. The code run AS IS to reproduce the issue.

Example Code

mycheckpointer = InMemorySaver()

class Orc:
    def __init__(self):
        self.agent = create_agent(
            model=register_llm(),
            checkpointer=mycheckpointer,
            system_prompt="Remember my name.",
        )

    def process(self, state):  # 🔥 类方法直接作 node！
        """直接作为 LangGraph node 函数。"""
        # print(state)
        config = {
            "configurable": {"thread_id": "1"},
        }
        state_response = self.agent.invoke(
            {'messages': [{'role': 'user', 'content': f"我是bob"}]},
            config={"configurable": {"thread_id": "1"}}

        )
        print(self.agent.get_state(config={"configurable": {"thread_id": "1"}}))
        # print(self.agent.get_state(config))
        state_response = self.agent.invoke(
            {'messages': [{'role': 'user', 'content': f"我是谁？"}]},
            config={"configurable": {"thread_id": "1"}}
        )
        print(state_response)
        print(self.agent.get_state(config={"configurable": {"thread_id": "1"}}))




if __name__ == "__main__":
    # 使用类方法作为 node (一行！)
    def hock(state):
        # print(state)
        state['name'] = 'alice'
    class Agent:
        def __init__(self):
            self.orc1 = Orc()
            self.orc2 = Orc()
            workflow = StateGraph(dict)
            workflow.add_node("orc1", self.orc1.process)  # 🔥 直接用类方法！
            # workflow.add_node("hock", hock)
            # workflow.add_node("orc2", self.orc2.process)  # 🔥 直接用类方法！
            workflow.add_edge(START, "orc1")
            # workflow.add_edge("orc1", 'hock')
            # workflow.add_edge("hock", 'orc2')
            workflow.add_edge("orc1", END)
            self.app = workflow.compile(checkpointer=mycheckpointer)


    myagent = Agent()
   
    for chunk in myagent.app.stream(
            {"name": 'bobbbb'},
            stream_mode="updates",
            subgraphs=True,
            config={"configurable": {"thread_id": "1"}}

    ):
        print(chunk)

Error Message and Stack Trace (if applicable)

Description

Overveiw

It is more like a usage oversight than a real error bug.
But I would like to express my requests, also my coding intentions (with this simple enough case).

Key findings

• By design or not, the langgraph is not compatible with langchain create_agent as a node.
• In original design, create_agent builds a graph instead of a node.
• However, with the rapid increase of multi-agent system, I would like to reusecreate_agent as a node. So that I can focus on state management in between.

Problems

• I want to capture the reasoning steps of the create_agent agent.
• while the streaming [update mode] of the langgraph support the updates of nodes, it cannot fetch those in create_agent .

1. I have tried not to use thread in create_agent. Luckily it works,

• However, in that way, the agent in each node cannot build its own checkpointer memory of messages.

2. I have tried to set thread for each agent to keep invoke memory in different outer langgraph round.

• However, in that way, the agent reasoning cannot be observed from outer agent.

This leads to a dead-end dilemma.

I am actually new to langchain, do you have any suggestions to build multi-agent systems (in a simpler way)? I am now so into your implementation of create_agent , rather than to build one myself.

System Info

just run with python, it has no error.
but Either you cannot see orc.agent memorizes between two invoke
or you cannot get subagent path of orc

[sambhavk]

• update stream mode only captures update made to langgraph state
• to capture reasoning steps of create_agent you have to first capture them in llm response and emit it as custom events
• Add custom in stream mode as well
• This is under assumption that reasoning steps means llm thinking which is emitted till llm generates final response

[EmmetWilliam]

I'd like to use an official demo to demonstrate my major concerns.

from langchain.agents import create_agent
from langgraph.graph import StateGraph, START, END

agent = create_agent(model="openai:gpt-4o", tools=[...])

def agent_node(state: State) -> dict:
    """A LangGraph node that invokes a LangChain agent."""
    result = agent.invoke({
        "messages": [{"role": "user", "content": state["query"]}]
    })
    return {"answer": result["messages"][-1].content}

# Build a simple workflow
workflow = (
    StateGraph(State)
    .add_node("agent", agent_node)
    .add_edge(START, "agent")
    .add_edge("agent", END)
    .compile()
)

In such a case, what can I do to let agent_node has its own short-term memory to maintain chat messages, while enabling its chat output in a stream mode.

• update stream mode only captures update made to langgraph state
• to capture reasoning steps of create_agent you have to first capture them in llm response and emit it as custom events
• Add custom in stream mode as well
• This is under assumption that reasoning steps means llm thinking which is emitted till llm generates final response

[EmmetWilliam]

I'd like to use an official demo to demonstrate my major concerns.

from langchain.agents import create_agent
from langgraph.graph import StateGraph, START, END

agent = create_agent(model="openai:gpt-4o", tools=[...])

def agent_node(state: State) -> dict:
    """A LangGraph node that invokes a LangChain agent."""
    result = agent.invoke({
        "messages": [{"role": "user", "content": state["query"]}]
    })
    return {"answer": result["messages"][-1].content}

# Build a simple workflow
workflow = (
    StateGraph(State)
    .add_node("agent", agent_node)
    .add_edge(START, "agent")
    .add_edge("agent", END)
    .compile()
)

In such a case, what can I do to let agent_node has its own short-term memory to maintain chat messages, while enabling its chat output in a stream mode.