Debugging quesiton: How to get a full traceback of where an error ocurred?

[hnykda]

Hi. I like langgraph, but one thing that I am struggling with is its debugging (in Python). For example, if some of my function returns some bad type with respect to my Pydantic model, I have no idea where in my 20 step workflow the error occurs.

Consider this output:

ValidationError                           Traceback (most recent call last)
Cell In[4], line 1
----> 1 await main.run_agent_loop(
      2     initial_state, agent, config, interrupt_before="collect_content"
      3 )

File ~/work/myproj/code/mypack/.venv/lib/python3.12/site-packages/langfuse/decorators/langfuse_decorator.py:217, in LangfuseDecorator._async_observe.<locals>.async_wrapper(*args, **kwargs)
    215     result = await func(*args, **kwargs)
    216 except Exception as e:
--> 217     self._handle_exception(observation, e)
    218 finally:
    219     result = self._finalize_call(
    220         observation, result, capture_output, transform_to_string
    221     )

File ~/work/myproj/code/mypack/.venv/lib/python3.12/site-packages/langfuse/decorators/langfuse_decorator.py:508, in LangfuseDecorator._handle_exception(self, observation, e)
    504 if observation:
    505     _observation_params_context.get()[observation.id].update(
    506         level="ERROR", status_message=str(e)
    507     )
--> 508 raise e

File ~/work/myproj/code/mypack/.venv/lib/python3.12/site-packages/langfuse/decorators/langfuse_decorator.py:215, in LangfuseDecorator._async_observe.<locals>.async_wrapper(*args, **kwargs)
    212 result = None
    214 try:
--> 215     result = await func(*args, **kwargs)
    216 except Exception as e:
    217     self._handle_exception(observation, e)

File ~/work/myproj/code/mypack/src/crawly/main.py:974, in run_agent_loop(state, agent, config, debug, **kwargs)
    965 @observe(name="Crawler", capture_output=False)
    966 async def run_agent_loop(
    967     state: ScrapingState,
   (...)
    971     **kwargs
    972 ):
    973     """Run the async agent loop"""
--> 974     async for event in agent.astream(state, config, stream_mode="updates", **kwargs):
    975         if debug:
    976             for node_name, _ in event.items():

File ~/work/myproj/code/mypack/.venv/lib/python3.12/site-packages/langgraph/pregel/__init__.py:1546, in Pregel.astream(self, input, config, stream_mode, output_keys, interrupt_before, interrupt_after, debug, subgraphs)
   1535 # Similarly to Bulk Synchronous Parallel / Pregel model
   1536 # computation proceeds in steps, while there are channel updates
   1537 # channel updates from step N are only visible in step N+1
   1538 # channels are guaranteed to be immutable for the duration of the step,
   1539 # with channel updates applied only at the transition between steps
   1540 while loop.tick(
   1541     input_keys=self.input_channels,
   1542     interrupt_before=interrupt_before_,
   1543     interrupt_after=interrupt_after_,
   1544     manager=run_manager,
   1545 ):
-> 1546     async for _ in runner.atick(
   1547         loop.tasks.values(),
   1548         timeout=self.step_timeout,
   1549         retry_policy=self.retry_policy,
   1550         get_waiter=get_waiter,
   1551     ):
   1552         # emit output
   1553         for o in output():
   1554             yield o

File ~/work/myproj/code/mypack/.venv/lib/python3.12/site-packages/langgraph/pregel/runner.py:132, in PregelRunner.atick(self, tasks, reraise, timeout, retry_policy, get_waiter)
    130 t = tasks[0]
    131 try:
--> 132     await arun_with_retry(t, retry_policy, stream=self.use_astream)
    133     self.commit(t, None)
    134 except Exception as exc:

File ~/work/myproj/code/mypack/.venv/lib/python3.12/site-packages/langgraph/pregel/retry.py:102, in arun_with_retry(task, retry_policy, stream)
    100         pass
    101 else:
--> 102     await task.proc.ainvoke(task.input, config)
    103 # if successful, end
    104 break

File ~/work/myproj/code/mypack/.venv/lib/python3.12/site-packages/langgraph/utils/runnable.py:453, in RunnableSeq.ainvoke(self, input, config, **kwargs)
    451     coro = step.ainvoke(input, config)
    452 if ASYNCIO_ACCEPTS_CONTEXT:
--> 453     input = await asyncio.create_task(coro, context=context)
    454 else:
    455     input = await asyncio.create_task(coro)

File ~/work/myproj/code/mypack/.venv/lib/python3.12/site-packages/langgraph/utils/runnable.py:236, in RunnableCallable.ainvoke(self, input, config, **kwargs)
    234 if ASYNCIO_ACCEPTS_CONTEXT:
    235     coro = cast(Coroutine[None, None, Any], self.afunc(input, **kwargs))
--> 236     ret = await asyncio.create_task(coro, context=context)
    237 else:
    238     ret = await self.afunc(input, **kwargs)

File ~/work/myproj/code/mypack/.venv/lib/python3.12/site-packages/langgraph/graph/graph.py:109, in Branch._aroute(self, input, config, reader, writer)
     98 async def _aroute(
     99     self,
    100     input: Any,
   (...)
    106     ],
    107 ) -> Runnable:
    108     if reader:
--> 109         value = await asyncio.to_thread(reader, config)
    110         # passthrough additional keys from node to branch
    111         # only doable when using dict states
    112         if isinstance(value, dict) and isinstance(input, dict):

File ~/.pyenv/versions/3.12.3/lib/python3.12/asyncio/threads.py:25, in to_thread(func, *args, **kwargs)
     23 ctx = contextvars.copy_context()
     24 func_call = functools.partial(ctx.run, func, *args, **kwargs)
---> 25 return await loop.run_in_executor(None, func_call)

File ~/.pyenv/versions/3.12.3/lib/python3.12/concurrent/futures/thread.py:58, in _WorkItem.run(self)
     55     return
     57 try:
---> 58     result = self.fn(*self.args, **self.kwargs)
     59 except BaseException as exc:
     60     self.future.set_exception(exc)

File ~/work/myproj/code/mypack/.venv/lib/python3.12/site-packages/langgraph/pregel/read.py:105, in ChannelRead.do_read(config, select, fresh, mapper)
    100     raise RuntimeError(
    101         "Not configured with a read function"
    102         "Make sure to call in the context of a Pregel process"
    103     )
    104 if mapper:
--> 105     return mapper(read(select, fresh))
    106 else:
    107     return read(select, fresh)

File ~/work/myproj/code/mypack/.venv/lib/python3.12/site-packages/langgraph/graph/state.py:743, in _coerce_state(schema, input)
    742 def _coerce_state(schema: Type[Any], input: dict[str, Any]) -> dict[str, Any]:
--> 743     return schema(**input)

File ~/work/myproj/code/mypack/.venv/lib/python3.12/site-packages/pydantic/main.py:212, in BaseModel.__init__(self, **data)
    210 # `__tracebackhide__` tells pytest and some other tools to omit this function from tracebacks
    211 __tracebackhide__ = True
--> 212 validated_self = self.__pydantic_validator__.validate_python(data, self_instance=self)
    213 if self is not validated_self:
    214     warnings.warn(
    215         'A custom validator is returning a value other than `self`.\n'
    216         "Returning anything other than `self` from a top level model validator isn't supported when validating via `__init__`.\n"
    217         'See the `model_validator` docs (https://docs.pydantic.dev/latest/concepts/validators/#model-validators) for more details.',
    218         category=None,
    219     )

ValidationError: 1 validation error for ScrapingState
knowledge_graph.0
  Input should be a valid dictionary [type=dict_type, input_value=<class 'dict'>, input_type=type]
    For further information visit https://errors.pydantic.dev/2.9/v/dict_type

Is there a way how I could make these errors coming from the framework and not my user code more legible, or knowing where does this happen? Would be great to show "Step X returned Y, which is incompatible with Z"

I tried e.g.:

        async for event in agent.astream(
            state, config, stream_mode="values", **kwargs
        ):
                for node_name, _ in event.items():
                    print(f"Executing node: {node_name}")

but that only shows everything when starting the agent, so hardly helpful:

Executing node: initial_query
Executing node: subqueries
Executing node: search_phrases
Executing node: people_also_ask
Executing node: related_searches
Executing node: organic_results
Executing node: knowledge_graph
Executing node: people_also_ask_filtered
Executing node: all_searches_filtered
Executing node: submitted_urls
Executing node: visited_pages_links