bisheng
bisheng copied to clipboard
dataelement
工作流总做了一个循环的流程，循环到第四次的时候会报错
[2025-04-22 20:45:25,553: WARNING/ForkPoolWorker-12] [2025-04-22 20:45:24.477534] [2025-04-22 20:45:24.300503] [ERROR process-323-139912903605120 bisheng.workflow.graph.graph_engine:290] - trace=40e5cc92593848c9ac8369d7d1040239 graph run error Traceback (most recent call last):
File "/usr/local/bin/celery", line 8, in <module>
sys.exit(main())
| | -> <function main at 0x7f4002bda560>
| -> <function Worker.__call__.<locals>.exit at 0x7f3f8e1d4790>
-> <module 'sys' (built-in)>
File "/usr/local/lib/python3.10/site-packages/celery/__main__.py", line 15, in main
sys.exit(_main())
| | -> <function main at 0x7f4000f69000>
| -> <function Worker.__call__.<locals>.exit at 0x7f3f8e1d4790>
-> <module 'sys' (built-in)>
File "/usr/local/lib/python3.10/site-packages/celery/bin/celery.py", line 236, in main
return celery(auto_envvar_prefix="CELERY")
-> <DYMGroup celery>
File "/usr/local/lib/python3.10/site-packages/click/core.py", line 1161, in __call__
return self.main(*args, **kwargs)
| | | -> {'auto_envvar_prefix': 'CELERY'}
| | -> ()
| -> <function BaseCommand.main at 0x7f40026a6440>
-> <DYMGroup celery>
File "/usr/local/lib/python3.10/site-packages/click/core.py", line 1082, in main
rv = self.invoke(ctx)
| | -> <click.core.Context object at 0x7f4002cf4460>
| -> <function MultiCommand.invoke at 0x7f40026a7400>
-> <DYMGroup celery>
File "/usr/local/lib/python3.10/site-packages/click/core.py", line 1697, in invoke
return _process_result(sub_ctx.command.invoke(sub_ctx))
| | | | -> <click.core.Context object at 0x7f4000f91090>
| | | -> <function Command.invoke at 0x7f40026a6ef0>
| | -> <CeleryDaemonCommand worker>
| -> <click.core.Context object at 0x7f4000f91090>
-> <function MultiCommand.invoke.<locals>._process_result at 0x7f4000f69240>
File "/usr/local/lib/python3.10/site-packages/click/core.py", line 1443, in invoke
return ctx.invoke(self.callback, **ctx.params)
| | | | | -> {'loglevel': 20, 'concurrency': 16, 'hostname': 'celery@65ba0276c1b7', 'detach': False, 'statedb': None, 'optimization': 'def...
| | | | -> <click.core.Context object at 0x7f4000f91090>
| | | -> <function worker at 0x7f4000f69ea0>
| | -> <CeleryDaemonCommand worker>
| -> <function Context.invoke at 0x7f40026a5cf0>
-> <click.core.Context object at 0x7f4000f91090>
File "/usr/local/lib/python3.10/site-packages/click/core.py", line 788, in invoke
return __callback(*args, **kwargs)
| -> {'loglevel': 20, 'concurrency': 16, 'hostname': 'celery@65ba0276c1b7', 'detach': False, 'statedb': None, 'optimization': 'def...
-> ()
File "/usr/local/lib/python3.10/site-packages/click/decorators.py", line 33, in new_func
return f(get_current_context(), *args, **kwargs)
| | | -> {'loglevel': 20, 'concurrency': 16, 'hostname': 'celery@65ba0276c1b7', 'detach': False, 'statedb': None, 'optimization': 'def...
| | -> ()
| -> <function get_current_context at 0x7f400267e440>
-> <function worker at 0x7f4000f69e10>
File "/usr/local/lib/python3.10/site-packages/celery/bin/base.py", line 135, in caller
return f(ctx, *args, **kwargs)
| | | -> {'loglevel': 20, 'concurrency': 16, 'hostname': 'celery@65ba0276c1b7', 'detach': False, 'statedb': None, 'optimization': 'def...
| | -> ()
| -> <click.core.Context object at 0x7f4000f91090>
-> <function worker at 0x7f4000f69d80>
File "/usr/local/lib/python3.10/site-packages/celery/bin/worker.py", line 356, in worker
worker.start()
| -> <function WorkController.start at 0x7f4000f15120>
-> <Worker: celery@65ba0276c1b7 (running)>
File "/usr/local/lib/python3.10/site-packages/celery/worker/worker.py", line 202, in start
self.blueprint.start(self)
| | | -> <Worker: celery@65ba0276c1b7 (running)>
| | -> <function Blueprint.start at 0x7f400119d750>
| -> <celery.worker.worker.WorkController.Blueprint object at 0x7f3f8e20f730>
-> <Worker: celery@65ba0276c1b7 (running)>
File "/usr/local/lib/python3.10/site-packages/celery/bootsteps.py", line 116, in start
step.start(parent)
| | -> <Worker: celery@65ba0276c1b7 (running)>
| -> <function StartStopStep.start at 0x7f400119e830>
-> <step: Pool>
File "/usr/local/lib/python3.10/site-packages/celery/bootsteps.py", line 365, in start
return self.obj.start()
| | -> <function BasePool.start at 0x7f4000f7dea0>
| -> <celery.concurrency.prefork.TaskPool object at 0x7f3f8e157f70>
-> <step: Pool>
File "/usr/local/lib/python3.10/site-packages/celery/concurrency/base.py", line 130, in start
self.on_start()
| -> <function TaskPool.on_start at 0x7f3f8e2683a0>
-> <celery.concurrency.prefork.TaskPool object at 0x7f3f8e157f70>
File "/usr/local/lib/python3.10/site-packages/celery/concurrency/prefork.py", line 109, in on_start
P = self._pool = Pool(processes=self.limit,
| | | | -> 16
| | | -> <celery.concurrency.prefork.TaskPool object at 0x7f3f8e157f70>
| | -> <class 'celery.concurrency.asynpool.AsynPool'>
| -> None
-> <celery.concurrency.prefork.TaskPool object at 0x7f3f8e157f70>
File "/usr/local/lib/python3.10/site-packages/celery/concurrency/asynpool.py", line 464, in __init__
super().__init__(processes, *args, **kwargs)
| | -> {'initializer': <function process_initializer at 0x7f3f8e31e830>, 'on_process_exit': <function process_destructor at 0x7f3f8e...
| -> ()
-> 16
File "/usr/local/lib/python3.10/site-packages/billiard/pool.py", line 1046, in __init__
self._create_worker_process(i)
| | -> 11
| -> <function AsynPool._create_worker_process at 0x7f3f8e212ef0>
-> <celery.concurrency.asynpool.AsynPool object at 0x7f3f8e259660>
File "/usr/local/lib/python3.10/site-packages/celery/concurrency/asynpool.py", line 482, in _create_worker_process
return super()._create_worker_process(i)
-> 11
File "/usr/local/lib/python3.10/site-packages/billiard/pool.py", line 1158, in _create_worker_process
w.start()
| -> <function BaseProcess.start at 0x7f4001e163b0>
-> <ForkProcess(ForkPoolWorker-12, started daemon)>
File "/usr/local/lib/python3.10/site-packages/billiard/process.py", line 120, in start
self._popen = self._Popen(self)
| | | | -> <ForkProcess(ForkPoolWorker-12, started daemon)>
| | | -> <staticmethod(<function ForkProcess._Popen at 0x7f4001e188b0>)>
| | -> <ForkProcess(ForkPoolWorker-12, started daemon)>
| -> None
-> <ForkProcess(ForkPoolWorker-12, started daemon)>
File "/usr/local/lib/python3.10/site-packages/billiard/context.py", line 331, in _Popen
return Popen(process_obj)
| -> <ForkProcess(ForkPoolWorker-12, started daemon)>
-> <class 'billiard.popen_fork.Popen'>
File "/usr/local/lib/python3.10/site-packages/billiard/popen_fork.py", line 22, in __init__
self._launch(process_obj)
| | -> <ForkProcess(ForkPoolWorker-12, started daemon)>
| -> <function Popen._launch at 0x7f3f8e1d44c0>
-> <billiard.popen_fork.Popen object at 0x7f3f8e206d10>
File "/usr/local/lib/python3.10/site-packages/billiard/popen_fork.py", line 77, in _launch
code = process_obj._bootstrap()
| -> <function BaseProcess._bootstrap at 0x7f4001e17010>
-> <ForkProcess(ForkPoolWorker-12, started daemon)>
File "/usr/local/lib/python3.10/site-packages/billiard/process.py", line 323, in _bootstrap
self.run()
| -> <function BaseProcess.run at 0x7f4001e16320>
-> <ForkProcess(ForkPoolWorker-12, started daemon)>
File "/usr/local/lib/python3.10/site-packages/billiard/process.py", line 110, in run
self._target(*self._args, **self._kwargs)
| | | | | -> {}
| | | | -> <ForkProcess(ForkPoolWorker-12, started daemon)>
| | | -> ()
| | -> <ForkProcess(ForkPoolWorker-12, started daemon)>
| -> <celery.concurrency.asynpool.Worker object at 0x7f3f8e206d70>
-> <ForkProcess(ForkPoolWorker-12, started daemon)>
File "/usr/local/lib/python3.10/site-packages/billiard/pool.py", line 292, in __call__
sys.exit(self.workloop(pid=pid))
| | | | -> 323
| | | -> <function Worker.workloop at 0x7f3f8e290430>
| | -> <celery.concurrency.asynpool.Worker object at 0x7f3f8e206d70>
| -> <function Worker.__call__.<locals>.exit at 0x7f3f8e1d4790>
-> <module 'sys' (built-in)>
File "/usr/local/lib/python3.10/site-packages/billiard/pool.py", line 362, in workloop
result = (True, prepare_result(fun(*args, **kwargs)))
| | | -> {}
| | -> ('bisheng.worker.workflow.tasks.execute_workflow', 'dc32a5c0-b08a-49cf-9867-2e5fa3c44fac', {'lang': 'py', 'task': 'bisheng.wo...
| -> <function fast_trace_task at 0x7f4000f88a60>
-> <bound method Worker.prepare_result of <celery.concurrency.asynpool.Worker object at 0x7f3f8e206d70>>
File "/usr/local/lib/python3.10/site-packages/celery/app/trace.py", line 651, in fast_trace_task
R, I, T, Rstr = tasks.__trace__(
| -> 'bisheng.worker.workflow.tasks.execute_workflow'
-> {'celery.group': <@task: celery.group of bisheng at 0x7f3fdae02170>, 'bisheng.worker.workflow.tasks.execute_workflow': <@task...
File "/usr/local/lib/python3.10/site-packages/celery/app/trace.py", line 453, in trace_task
R = retval = fun(*args, **kwargs)
| | -> {}
| -> ['40e5cc92593848c9ac8369d7d1040239', 'e3184a1d-d0f8-4d03-a3a5-80cb7275ba07', '', '1']
-> <@task: bisheng.worker.workflow.tasks.execute_workflow of bisheng at 0x7f3fdae02170>
File "/usr/local/lib/python3.10/site-packages/celery/app/trace.py", line 736, in __protected_call__
return self.run(*args, **kwargs)
| | | -> {}
| | -> ('40e5cc92593848c9ac8369d7d1040239', 'e3184a1d-d0f8-4d03-a3a5-80cb7275ba07', '', '1')
| -> <staticmethod(<function execute_workflow at 0x7f3fa80f3010>)>
-> <@task: bisheng.worker.workflow.tasks.execute_workflow of bisheng at 0x7f3fdae02170>
File "/app/bisheng/worker/workflow/tasks.py", line 69, in execute_workflow
_execute_workflow(unique_id, workflow_id, chat_id, user_id)
| | | | -> '1'
| | | -> ''
| | -> 'e3184a1d-d0f8-4d03-a3a5-80cb7275ba07'
| -> '40e5cc92593848c9ac8369d7d1040239'
-> <function _execute_workflow at 0x7f3fa80f2f80>
File "/app/bisheng/worker/workflow/tasks.py", line 53, in _execute_workflow
status, reason = workflow.run(user_input)
| | | -> {'output_f1084': {'output_result': '92463cb1'}}
| | -> <function Workflow.run at 0x7f3fa3f99e10>
| -> <bisheng.workflow.graph.workflow.Workflow object at 0x7f3fd97b0e20>
-> 'INPUT'
File "/app/bisheng/workflow/graph/workflow.py", line 63, in run
self.graph_engine.continue_run()
| | -> <function GraphEngine.continue_run at 0x7f3fa3f99b40>
| -> <bisheng.workflow.graph.graph_engine.GraphEngine object at 0x7f3f8e2078b0>
-> <bisheng.workflow.graph.workflow.Workflow object at 0x7f3fd97b0e20>
File "/app/bisheng/workflow/graph/graph_engine.py", line 330, in continue_run
self._run(None)
| -> <function GraphEngine._run at 0x7f3fa3f99900>
-> <bisheng.workflow.graph.graph_engine.GraphEngine object at 0x7f3f8e2078b0>
> File "/app/bisheng/workflow/graph/graph_engine.py", line 282, in _run
for _ in self.graph.stream(input_data, config=self.graph_config):
| | | | | | -> {'configurable': {'thread_id': '1'}, 'recursion_limit': 18000}
| | | | | -> <bisheng.workflow.graph.graph_engine.GraphEngine object at 0x7f3f8e2078b0>
| | | | -> None
| | | -> <function Pregel.stream at 0x7f3fa81ace50>
| | -> <langgraph.graph.state.CompiledStateGraph object at 0x7f3f8e155300>
| -> <bisheng.workflow.graph.graph_engine.GraphEngine object at 0x7f3f8e2078b0>
-> {'code_d6500': {'flag': 23346660468288651264}}
File "/usr/local/lib/python3.10/site-packages/langgraph/pregel/__init__.py", line 1937, in stream
with SyncPregelLoop(
-> <class 'langgraph.pregel.loop.SyncPregelLoop'>
File "/usr/local/lib/python3.10/site-packages/langgraph/pregel/loop.py", line 963, in __exit__
return self.stack.__exit__(exc_type, exc_value, traceback)
| | | | | -> <traceback object at 0x7f3fcba7c2c0>
| | | | -> GraphInterrupt(())
| | | -> <class 'langgraph.errors.GraphInterrupt'>
| | -> <function ExitStack.__exit__ at 0x7f4002b48160>
| -> <contextlib.ExitStack object at 0x7f3ffda6ecb0>
-> <langgraph.pregel.loop.SyncPregelLoop object at 0x7f3ffda44f70>
File "/usr/local/lib/python3.10/contextlib.py", line 576, in __exit__
raise exc_details[1]
-> (<class 'TypeError'>, TypeError('Object of type int is not serializable'), <traceback object at 0x7f3faa95f5c0>)
File "/usr/local/lib/python3.10/contextlib.py", line 561, in __exit__
if cb(*exc_details):
| -> (<class 'TypeError'>, TypeError('Object of type int is not serializable'), <traceback object at 0x7f3faa95f5c0>)
-> <bound method BackgroundExecutor.__exit__ of <langgraph.pregel.executor.BackgroundExecutor object at 0x7f3ffda6f3a0>>
File "/usr/local/lib/python3.10/site-packages/langgraph/pregel/executor.py", line 120, in __exit__
task.result()
| -> <function Future.result at 0x7f40018a0940>
-> <Future at 0x7f3ffda467d0 state=finished raised TypeError>
File "/usr/local/lib/python3.10/concurrent/futures/_base.py", line 451, in result
return self.__get_result()
-> None
File "/usr/local/lib/python3.10/concurrent/futures/_base.py", line 403, in __get_result
raise self._exception
-> None
File "/usr/local/lib/python3.10/site-packages/langgraph/pregel/loop.py", line 884, in _checkpointer_put_after_previous
prev.result()
| -> <function Future.result at 0x7f40018a0940>
-> <Future at 0x7f3ffda467d0 state=finished raised TypeError>
File "/usr/local/lib/python3.10/concurrent/futures/_base.py", line 451, in result
return self.__get_result()
-> None
File "/usr/local/lib/python3.10/concurrent/futures/_base.py", line 403, in __get_result
raise self._exception
-> None
File "/usr/local/lib/python3.10/site-packages/langgraph/pregel/executor.py", line 83, in done
task.result()
| -> <function Future.result at 0x7f40018a0940>
-> <Future at 0x7f3ffda467d0 state=finished raised TypeError>
File "/usr/local/lib/python3.10/concurrent/futures/_base.py", line 451, in result
return self.__get_result()
-> None
File "/usr/local/lib/python3.10/concurrent/futures/_base.py", line 403, in __get_result
raise self._exception
-> None
File "/usr/local/lib/python3.10/concurrent/futures/thread.py", line 58, in run
result = self.fn(*self.args, **self.kwargs)
| | -> None
| -> None
-> None
File "/usr/local/lib/python3.10/site-packages/langgraph/pregel/loop.py", line 886, in _checkpointer_put_after_previous
cast(BaseCheckpointSaver, self.checkpointer).put(
| | | -> <langgraph.checkpoint.memory.InMemorySaver object at 0x7f3f8e134d00>
| | -> <langgraph.pregel.loop.SyncPregelLoop object at 0x7f3ffda44f70>
| -> <class 'langgraph.checkpoint.base.BaseCheckpointSaver'>
-> <function cast at 0x7f40029a7eb0>
File "/usr/local/lib/python3.10/site-packages/langgraph/checkpoint/memory/__init__.py", line 359, in put
self.serde.dumps_typed(c),
| | | -> {'v': 1, 'ts': '2025-04-22T12:45:24.227775+00:00', 'id': '1f01f77a-8d05-65c9-803a-c22c1d464438', 'channel_values': {'flag': 2...
| | -> <function JsonPlusSerializer.dumps_typed at 0x7f3fa8149ab0>
| -> <langgraph.checkpoint.serde.jsonplus.JsonPlusSerializer object at 0x7f3fa9098fa0>
-> <langgraph.checkpoint.memory.InMemorySaver object at 0x7f3f8e134d00>
File "/usr/local/lib/python3.10/site-packages/langgraph/checkpoint/serde/jsonplus.py", line 194, in dumps_typed
return "msgpack", _msgpack_enc(obj)
| -> {'v': 1, 'ts': '2025-04-22T12:45:24.227775+00:00', 'id': '1f01f77a-8d05-65c9-803a-c22c1d464438', 'channel_values': {'flag': 2...
-> <function _msgpack_enc at 0x7f3fa8149d80>
File "/usr/local/lib/python3.10/site-packages/langgraph/checkpoint/serde/jsonplus.py", line 507, in _msgpack_enc
return msgpack.packb(data, default=_msgpack_default)
| | | -> <function _msgpack_default at 0x7f3fa82decb0>
| | -> {'v': 1, 'ts': '2025-04-22T12:45:24.227775+00:00', 'id': '1f01f77a-8d05-65c9-803a-c22c1d464438', 'channel_values': {'flag': 2...
| -> <function packb at 0x7f4001d61510>
-> <module 'msgpack' from '/usr/local/lib/python3.10/site-packages/msgpack/__init__.py'>
File "/usr/local/lib/python3.10/site-packages/msgpack/__init__.py", line 36, in packb
return Packer(**kwargs).pack(o)
| | -> {'v': 1, 'ts': '2025-04-22T12:45:24.227775+00:00', 'id': '1f01f77a-8d05-65c9-803a-c22c1d464438', 'channel_values': {'flag': 2...
| -> {'default': <function _msgpack_default at 0x7f3fa82decb0>}
-> <class 'msgpack._cmsgpack.Packer'>
File "msgpack/_packer.pyx", line 279, in msgpack._cmsgpack.Packer.pack
File "msgpack/_packer.pyx", line 276, in msgpack._cmsgpack.Packer.pack
File "msgpack/_packer.pyx", line 265, in msgpack._cmsgpack.Packer._pack
File "msgpack/_packer.pyx", line 213, in msgpack._cmsgpack.Packer._pack_inner
File "msgpack/_packer.pyx", line 265, in msgpack._cmsgpack.Packer._pack
File "msgpack/_packer.pyx", line 213, in msgpack._cmsgpack.Packer._pack_inner
File "msgpack/_packer.pyx", line 267, in msgpack._cmsgpack.Packer._pack
File "/usr/local/lib/python3.10/site-packages/langgraph/checkpoint/serde/jsonplus.py", line 436, in _msgpack_default
raise TypeError(f"Object of type {obj.__class__.__name__} is not serializable")
TypeError: Object of type int is not serializable
