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

intermediate_steps in playground not complete after CompiledGraph invoke #157

New Issue
Closed
opened 2026-02-16 00:19:02 -05:00 by yindo · 1 comment
yindo commented 2026-02-16 00:19:02 -05:00
Owner
Copy Link
Copy Source

Originally created by @weiminw on GitHub (Apr 7, 2024).

import operator
from typing import TypedDict, Union, Annotated, List, Any, Optional

from langchain.agents import create_structured_chat_agent
from langchain_community.tools.ddg_search import DuckDuckGoSearchRun
from langchain_core.agents import AgentAction, AgentFinish
from langchain_core.messages import BaseMessage
from langchain_core.runnables import ConfigurableField, RunnablePassthrough, RunnableLambda, Runnable, RunnableConfig
from langchain_core.runnables.utils import Input, Output, ConfigurableFieldSpec, AddableDict
from langgraph.graph import StateGraph, END
from langgraph.graph.graph import CompiledGraph, Graph
from langgraph.prebuilt import ToolExecutor
from pydantic.v1 import BaseModel, PrivateAttr

from heliumos.jarvis import AgentConversationInput, llm, AgentConversationOutput
from heliumos.jarvis.prompts.heliumos_react_prompt import CharacterPrompt, prompt


class AgentState(TypedDict):
    # The input string
    input: str
    # The list of previous messages in the conversation
    chat_history: list[BaseMessage]
    # The outcome of a given call to the agent
    # Needs `None` as a valid type, since this is what this will start as
    agent_outcome: Union[AgentAction, AgentFinish, None]
    # List of actions and corresponding observations
    # Here we annotate this with `operator.add` to indicate that operations to
    # this state should be ADDED to the existing values (not overwrite it)
    intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add]

def parse_output(out: Any):
    return {"output": out["agent_outcome"].return_values["output"],
            "intermediate_steps": out["intermediate_steps"]}


class HeliumOSReActGraphAgent(Runnable, BaseModel):
    workflow: Graph
    agent_runnable: Runnable
    tool_executor: ToolExecutor
    _workflow_complied: Optional[CompiledGraph] = PrivateAttr()

    class Config:
        arbitrary_types_allowed = True

    def _run_agent(self, state: AgentState):
        agent_outcome = self.agent_runnable.invoke(
            {"input": state["input"], "intermediate_steps": state["intermediate_steps"]}
        )
        return {"agent_outcome": agent_outcome}

    def _execute_tools(self, state: AgentState):
        # Get the most recent agent_outcome - this is the key added in the `agent` above
        agent_action = state["agent_outcome"]
        output = self.tool_executor.invoke(agent_action)
        return {"intermediate_steps": [(agent_action, str(output))]}

    def _should_continue(self, data):
        # If the agent outcome is an AgentFinish, then we return `exit` string
        # This will be used when setting up the graph to define the flow
        if isinstance(data["agent_outcome"], AgentFinish):
            return "end"
        # Otherwise, an AgentAction is returned
        # Here we return `continue` string
        # This will be used when setting up the graph to define the flow
        else:
            return "continue"
    @property
    def config_specs(self) -> List[ConfigurableFieldSpec]:
        return self.agent_runnable.config_specs

    def compile(self) -> Runnable:
        self.workflow.add_node("agent", self._run_agent)
        self.workflow.add_node("action", self._execute_tools)
        # Set the entrypoint as `agent`
        # This means that this node is the first one called
        self.workflow.set_entry_point("agent")
        # We now add a conditional edge
        self.workflow.add_conditional_edges(
            # First, we define the start node. We use `agent`.
            # This means these are the edges taken after the `agent` node is called.
            "agent",
            # Next, we pass in the function that will determine which node is called next.
            self._should_continue,
            # Finally we pass in a mapping.
            # The keys are strings, and the values are other nodes.
            # END is a special node marking that the graph should finish.
            # What will happen is we will call `should_continue`, and then the output of that
            # will be matched against the keys in this mapping.
            # Based on which one it matches, that node will then be called.
            {
                # If `tools`, then we call the tool node.
                "continue": "action",
                # Otherwise we finish.
                "end": END,
            },
        )

        # We now add a normal edge from `tools` to `agent`.
        # This means that after `tools` is called, `agent` node is called next.
        self.workflow.add_edge("action", "agent")

        # Finally, we compile it!
        # This compiles it into a LangChain Runnable,
        # meaning you can use it as you would any other runnable
        self._workflow_complied = self.workflow.compile()
        return self
    def invoke(self, input: Input, config: Optional[RunnableConfig] = None) -> Output:

        response = self._workflow_complied.invoke({"input": input["input"], "chat_history": [],})
        return response


def create_react_graph() -> Runnable:
    character_prompt = CharacterPrompt().configurable_fields(
        character=ConfigurableField(
            id="character"
        ),
    )

    agent_runnable = RunnablePassthrough.assign(character=character_prompt) | create_structured_chat_agent(
        llm=llm,
        tools=[DuckDuckGoSearchRun()],
        prompt=prompt,
        stop_sequence=["\nObserv", "\nObservation"], )

    agent_with_output_parser = HeliumOSReActGraphAgent(
        workflow=StateGraph(AgentState),
        agent_runnable=agent_runnable,
        tool_executor=ToolExecutor([DuckDuckGoSearchRun()])
    ).compile().with_types(
        input_type=AgentConversationInput,
        output_type=AgentConversationOutput
    ) | RunnableLambda(parse_output)

    return agent_with_output_parser


....
react_graph = create_react_graph()
add_routes(
    app,
    react_graph,
playground_type="default",
    path="/test"
)

I test input with " search new movies in 2024" .
I found no intermediate_steps in playground return to me .

Originally created by @weiminw on GitHub (Apr 7, 2024). ``` import operator from typing import TypedDict, Union, Annotated, List, Any, Optional from langchain.agents import create_structured_chat_agent from langchain_community.tools.ddg_search import DuckDuckGoSearchRun from langchain_core.agents import AgentAction, AgentFinish from langchain_core.messages import BaseMessage from langchain_core.runnables import ConfigurableField, RunnablePassthrough, RunnableLambda, Runnable, RunnableConfig from langchain_core.runnables.utils import Input, Output, ConfigurableFieldSpec, AddableDict from langgraph.graph import StateGraph, END from langgraph.graph.graph import CompiledGraph, Graph from langgraph.prebuilt import ToolExecutor from pydantic.v1 import BaseModel, PrivateAttr from heliumos.jarvis import AgentConversationInput, llm, AgentConversationOutput from heliumos.jarvis.prompts.heliumos_react_prompt import CharacterPrompt, prompt class AgentState(TypedDict): # The input string input: str # The list of previous messages in the conversation chat_history: list[BaseMessage] # The outcome of a given call to the agent # Needs `None` as a valid type, since this is what this will start as agent_outcome: Union[AgentAction, AgentFinish, None] # List of actions and corresponding observations # Here we annotate this with `operator.add` to indicate that operations to # this state should be ADDED to the existing values (not overwrite it) intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add] def parse_output(out: Any): return {"output": out["agent_outcome"].return_values["output"], "intermediate_steps": out["intermediate_steps"]} class HeliumOSReActGraphAgent(Runnable, BaseModel): workflow: Graph agent_runnable: Runnable tool_executor: ToolExecutor _workflow_complied: Optional[CompiledGraph] = PrivateAttr() class Config: arbitrary_types_allowed = True def _run_agent(self, state: AgentState): agent_outcome = self.agent_runnable.invoke( {"input": state["input"], "intermediate_steps": state["intermediate_steps"]} ) return {"agent_outcome": agent_outcome} def _execute_tools(self, state: AgentState): # Get the most recent agent_outcome - this is the key added in the `agent` above agent_action = state["agent_outcome"] output = self.tool_executor.invoke(agent_action) return {"intermediate_steps": [(agent_action, str(output))]} def _should_continue(self, data): # If the agent outcome is an AgentFinish, then we return `exit` string # This will be used when setting up the graph to define the flow if isinstance(data["agent_outcome"], AgentFinish): return "end" # Otherwise, an AgentAction is returned # Here we return `continue` string # This will be used when setting up the graph to define the flow else: return "continue" @property def config_specs(self) -> List[ConfigurableFieldSpec]: return self.agent_runnable.config_specs def compile(self) -> Runnable: self.workflow.add_node("agent", self._run_agent) self.workflow.add_node("action", self._execute_tools) # Set the entrypoint as `agent` # This means that this node is the first one called self.workflow.set_entry_point("agent") # We now add a conditional edge self.workflow.add_conditional_edges( # First, we define the start node. We use `agent`. # This means these are the edges taken after the `agent` node is called. "agent", # Next, we pass in the function that will determine which node is called next. self._should_continue, # Finally we pass in a mapping. # The keys are strings, and the values are other nodes. # END is a special node marking that the graph should finish. # What will happen is we will call `should_continue`, and then the output of that # will be matched against the keys in this mapping. # Based on which one it matches, that node will then be called. { # If `tools`, then we call the tool node. "continue": "action", # Otherwise we finish. "end": END, }, ) # We now add a normal edge from `tools` to `agent`. # This means that after `tools` is called, `agent` node is called next. self.workflow.add_edge("action", "agent") # Finally, we compile it! # This compiles it into a LangChain Runnable, # meaning you can use it as you would any other runnable self._workflow_complied = self.workflow.compile() return self def invoke(self, input: Input, config: Optional[RunnableConfig] = None) -> Output: response = self._workflow_complied.invoke({"input": input["input"], "chat_history": [],}) return response def create_react_graph() -> Runnable: character_prompt = CharacterPrompt().configurable_fields( character=ConfigurableField( id="character" ), ) agent_runnable = RunnablePassthrough.assign(character=character_prompt) | create_structured_chat_agent( llm=llm, tools=[DuckDuckGoSearchRun()], prompt=prompt, stop_sequence=["\nObserv", "\nObservation"], ) agent_with_output_parser = HeliumOSReActGraphAgent( workflow=StateGraph(AgentState), agent_runnable=agent_runnable, tool_executor=ToolExecutor([DuckDuckGoSearchRun()]) ).compile().with_types( input_type=AgentConversationInput, output_type=AgentConversationOutput ) | RunnableLambda(parse_output) return agent_with_output_parser .... react_graph = create_react_graph() add_routes( app, react_graph, playground_type="default", path="/test" ) ``` I test input with " search new movies in 2024" . I found no intermediate_steps in playground return to me .
yindo closed this issue 2026-02-16 00:19:03 -05:00
yindo commented 2026-02-16 00:19:03 -05:00
Author
Owner
Copy Link
Copy Source

@weiminw commented on GitHub (Apr 8, 2024):

resolved， upgrade langgraph to 0.0.32 and add CompliedGraph with .with_config(config)

@weiminw commented on GitHub (Apr 8, 2024): resolved， upgrade langgraph to 0.0.32 and add CompliedGraph with .with_config(config)
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#157
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?