KeyError encountered: '__start__'

[anshulbhide]

I’m encountering a KeyError: 'start' when invoking a runnable graph in my Langchain-based agent. This error seems to occur when the graph tries to access a start key during execution. Additionally, due to this error, the output (out) is not defined, which leads to a subsequent NameError.

Error Traceback:

Traceback (most recent call last):
File "/home/runner/langchain-agent-profile-viewer/main.py", line 296, in
out = runnable.invoke({
File "/home/runner/langchain-agent-profile-viewer/.pythonlibs/lib/python3.10/site-packages/langgraph/pregel/init.py", line 1451, in invoke
for chunk in self.stream(
File "/home/runner/langchain-agent-profile-viewer/.pythonlibs/lib/python3.10/site-packages/langgraph/pregel/init.py", line 898, in stream
next_checkpoint, next_tasks = _prepare_next_tasks(
File "/home/runner/langchain-agent-profile-viewer/.pythonlibs/lib/python3.10/site-packages/langgraph/pregel/init.py", line 1772, in _prepare_next_tasks
seen = checkpoint["versions_seen"][name]
KeyError: 'start'

Context
I am using Langchain to build an agent with several tools (linkedin_search, web_search, and final_answer).
The error occurs when invoking the compiled runnable graph using the invoke method.
The graph uses an initial state that includes the input, chat_history, and intermediate_steps keys. The start key is intended to trigger the start of the graph, but it seems the graph is not handling this key properly.

Questions:

1. Is there a specific way to ensure that the start key is correctly managed in a Langchain runnable graph?
2. How can I better structure the initial state or the graph configuration to avoid this KeyError?
3. What are the best practices for handling such initialization issues in Langchain, especially when using the start key?

Relevant code
initial_state = {
"start": "initial_value", # Adjusted as needed
"input": "https://www.linkedin.com/in/anshul-bhide-26a96026/",
"chat_history": [],
"intermediate_steps": [],
}

print("Initial state:", initial_state)

try:
out = runnable.invoke(initial_state)
print("Output:", out)
except KeyError as e:
print(f"KeyError encountered: {e}")
out = None
except Exception as e:
print(f"An unexpected error occurred: {e}")
out = None

if out is not None:
try:
final_report = build_report(output=out["intermediate_steps"][-1].tool_input)
print(final_report)
except NameError as e:
print(f"NameError encountered: {e}")
else:
print("No valid output available to build the report.")

[hwchase17]

hmm are you able to share the code for this graph?

[anshulbhide]

Here's the code -

import os
import streamlit as st
from langchain.agents import Tool
from langchain_core.tools import tool
from langgraph.graph import StateGraph, END

Removed API key environment variable setup

Initialize agent state

from typing import TypedDict, Annotated
from langchain_core.agents import AgentAction
from langchain_core.messages import BaseMessage
import operator

class AgentState(TypedDict):
input: str
chat_history: list[BaseMessage]
intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add]

Define search tool

from serpapi import GoogleSearch

@tool("web_search")
def web_search(name: str):
"""Finds general knowledge information using Google search. Can also be used
to augment more 'general' knowledge to a previous specialist query."""
serpapi_params = {
"engine": "google",
"api_key": "your_serpapi_key_here"
}
search = GoogleSearch({
**serpapi_params,
"q": name
})
results = search.get_dict()
organic_results = results.get("organic_results", [])

# Initialize an empty list to hold each context string
contexts = []

# Loop through the results and format each one
for result in organic_results:
    title = result.get("title", "No Title")
    snippet = result.get("snippet", "No Description")
    link = result.get("link", "No Link")

    # Format and add each context to the list
    context = f"{title}\n{snippet}\n{link}\n---"
    contexts.append(context)

# Join all contexts into a single string
final_context = "\n".join(contexts)

return final_context

Define final answer tool

@tool("final_answer")
def final_answer(
introduction: str,
research_steps: str,
main_body: str,
conclusion: str,
sources: str
):
"""Returns a natural language response to the user in the form of a research
report on the prospect's target. There are several sections to this report, those are:

• introduction: a short paragraph introducing the user's question and the
  topic we are researching.
• research_steps: a few bullet points explaining the steps that were taken
  to research your report.
• main_body: this is where the bulk of high quality and concise
  information that answers the user's question belongs. It is 3-4 paragraphs
  long in length.
• conclusion: this is a short single paragraph conclusion providing a
  concise but sophisticated view on what was found.
• sources: a bulletpoint list provided detailed sources for all information
  referenced during the research process
  """
  if type(research_steps) is list:
  research_steps = "\n".join([f"- {r}" for r in research_steps])
  if type(sources) is list:
  sources = "\n.join([f"- {s}" for s in sources])
  return ""

Create an oracle

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

system_prompt = """You are an excellent research assistant. You are tasked with searching a person's profile and have to research information on the person in a sales context. You will use web search tools for this purpose."""

prompt = ChatPromptTemplate.from_messages([
("system", system_prompt),
MessagesPlaceholder(variable_name="chat_history"),
("user", "{input}"),
("assistant", "scratchpad: {scratchpad}"),
])

from langchain_core.messages import ToolCall, ToolMessage
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
model="gpt-4o",
temperature=0.2
)

tools = [
web_search
]

Define a function to transform intermediate_steps from list

of AgentAction to scratchpad string

def create_scratchpad(intermediate_steps: list[AgentAction]):
research_steps = []
for i, action in enumerate(intermediate_steps):
if action.log != "TBD":
research_steps.append(
f"Tool: {action.tool}, input: {action.tool_input}\n"
f"Output: {action.log}"
)
return "\n---\n".join(research_steps)

oracle = (
{
"input": lambda x: x["input"],
"chat_history": lambda x: x["chat_history"],
"scratchpad": lambda x: create_scratchpad(
intermediate_steps=x["intermediate_steps"]
),
}
| prompt
| llm.bind_tools(tools, tool_choice="any")
)

inputs = {
"input": "example_input_data",
"chat_history": [],
"intermediate_steps": [],
}

Define nodes for graph

def run_oracle(state: list):
print("run_oracle")
intermediate_steps = state.get('intermediate_steps', [])
print(f"intermediate_steps: {intermediate_steps}")

out = oracle.invoke(state)

if out.tool_calls and len(out.tool_calls) > 0:
    tool_name = out.tool_calls[0].get("name")
    tool_args = out.tool_calls[0].get("args")
else:
    raise ValueError("No tool calls were made by the oracle.")

action_out = AgentAction(
    tool=tool_name,
    tool_input=tool_args,
    log="TBD"
)

intermediate_steps.append(action_out)

return {
    "intermediate_steps": intermediate_steps
}

def router(state: list):
# return the tool name to use
if isinstance(state["intermediate_steps"], list):
return state["intermediate_steps"][-1].tool
else:
# if we output bad format go to final answer
print("Router invalid format")
return "final_answer"

tool_str_to_func = {
"web_search": web_search,
"final_answer": final_answer
}

def run_tool(state: list):
tool_name = state["intermediate_steps"][-1].tool
tool_args = state["intermediate_steps"][-1].tool_input
print(f"{tool_name}.invoke(input={tool_args})")
out = tool_str_to_func[tool_name].invoke(input=tool_args)
action_out = AgentAction(
tool=tool_name,
tool_input=tool_args,
log=str(out)
)
return {"intermediate_steps": [action_out]}

Define graph

graph = StateGraph(AgentState)

graph.add_node("oracle", run_oracle)
graph.add_node("web_search", run_tool)
graph.add_node("final_answer", run_tool)

graph.set_entry_point("oracle")

graph.add_conditional_edges(
source="oracle", # where in graph to start
path=router, # function to determine which node is called
)

Create edges from each tool back to the oracle

for tool_obj in tools:
if tool_obj.name != "final_answer":
graph.add_edge(tool_obj.name, "oracle")

If anything goes to final answer, it must then move to END

graph.add_edge("final_answer", END)

runnable = graph.compile()

initial_state = {
"start": "oracle",
"input": "example_input_data",
"chat_history": [],
"intermediate_steps": [],
}

out = runnable.invoke(initial_state)
print(out)

def build_report(output: dict):
research_steps = output["research_steps"]
if type(research_steps) is list:
research_steps = "\n".join([f"- {r}" for r in research_steps])
sources = output["sources"]
if type(sources) is list:
sources = "\n".join([f"- {s}" for s in sources])
return f"""
INTRODUCTION

{output["introduction"]}

RESEARCH STEPS

{research_steps}

REPORT

{output["main_body"]}

CONCLUSION

{output["conclusion"]}

SOURCES

{sources}
"""

print(build_report(
output=out["intermediate_steps"][-1].tool_input
))

[vbarda]

hm, which version of langgraph are you using? i couldn't reproduce the error you posted

[anshulbhide]

Version: 0.1.1

[anshulbhide]

Okay so it worked when I updated to the latest version langgraph-0.2.14 - thanks!