ERROR:llama_deploy.services.workflow - Encountered error in task 24576e5e-edff-49cc-b3bf-55008826d818! keys must be str, int, float, bool or None, not ModelMetaclass #148

New Issue

Closed

opened 2026-02-16 01:16:01 -05:00 by yindo · 5 comments

yindo commented 2026-02-16 01:16:01 -05:00

Owner

Copy Link

Copy Source

Originally created by @ifreeman6 on GitHub (Nov 27, 2024).

When I run python -m llama_deploy.apiserver and deploy my workflow, I request the API from front-end [/deployments/{deployment_name}/tasks/{task_id}/events], occurs the following errors in back-end:

ERROR:llama_deploy.services.workflow - Encountered error in task 24576e5e-edff-49cc-b3bf-55008826d818! keys must be str, int, float, bool or None, not ModelMetaclass

I try different methods, still not works...
help me. Thanks very much!

This is my workflow

import json
from llama_index.core.workflow import (
    step,
    Context,
    Workflow,
    Event,
    StartEvent,
    StopEvent,
)
from llama_index.core.agent import ReActAgent
from llama_index.core.settings import Settings
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
import os
from llama_index.core import SimpleDirectoryReader, VectorStoreIndex, load_index_from_storage, StorageContext
from llama_index.core.base.base_query_engine import BaseQueryEngine
from llama_index.core.tools import FunctionTool
from datetime import datetime
from dotenv import load_dotenv
import asyncio
load_dotenv()

class QueryEvent(Event):
    question: str

class AnswerEvent(Event):
    question: str
    answer: str

class ProgressEvent(Event):
    def __init__(self, status: str, message: str, timestamp: str = None):
        super().__init__()
        self.metadata = {
            "progress": {
                "status": status,
                "message": message,
                "timestamp": timestamp or datetime.now().isoformat()
            }
        }

class SubQuestionQueryEngine(Workflow):
    def __init__(self, query_engine: BaseQueryEngine, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.query_engine = query_engine
        self.llm = OpenAI(
            api_base=os.getenv("OPENAI_API_BASE"),
            api_key=os.getenv("OPENAI_API_KEY"),
            model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"),
            timeout=120,
            max_retries=3
        )
        self.default_tools = [
            {
                "name": "search",
                "description": "Search through the document database to find relevant information"
            }
        ]

    @step
    async def query(self, ctx: Context, ev: StartEvent) -> QueryEvent:
        try:
            if hasattr(ev, "query"):
                await ctx.set("original_query", ev.query)
                print(f"Query is {await ctx.get('original_query')}")

            await ctx.set("llm", self.llm)
            
            tools = self.default_tools
            if hasattr(ev, "tools"):
                tools = ev.tools
            await ctx.set("tools", tools)

            # 发送初始进度事件
            ctx.write_event_to_stream(ProgressEvent(
                status="starting",
                message="Starting query analysis"
            ))

            # 修改这里的工具描述获��方式
            tool_descriptions = [tool["description"] for tool in tools]
            
            response = (await ctx.get("llm")).complete(
                f"""
                Given a user question, and a list of tools, output a list of
                relevant sub-questions (not more than 3), such that the answers to all the
                sub-questions put together will answer the question. Respond
                in pure JSON without any markdown, like this:
                {{
                    "sub_questions": [
                        "What is the population of San Francisco?",
                        "What is the budget of San Francisco?",
                        "What is the GDP of San Francisco?"
                    ]
                }}
                Here is the user question: {await ctx.get('original_query')}

                Available tools:
                {tool_descriptions}
                """
            )

            print(f"Sub-questions are {response}")

            response_obj = json.loads(str(response))
            sub_questions = response_obj["sub_questions"]

            await ctx.set("sub_question_count", len(sub_questions))

            # 发送进度事件
            ctx.write_event_to_stream(ProgressEvent(
                status="running",
                message=f"Generated {len(sub_questions)} sub-questions"
            ))

            # 逐个发送子问题
            for question in sub_questions:
                ctx.send_event(QueryEvent(question=question))

            return None

        except Exception as e:
            ctx.write_event_to_stream(ProgressEvent(
                status="error",
                message=str(e)
            ))
            raise

    @step
    async def sub_question(self, ctx: Context, ev: QueryEvent) -> AnswerEvent:
        try:
            ctx.write_event_to_stream(ProgressEvent(
                status="processing",
                message=f"Processing sub-question: {ev.question}"
            ))

            print(f"Sub-question is {ev.question}")

            tools = await ctx.get("tools")
            agent_tools = [
                FunctionTool.from_defaults(
                    fn=self.query_engine.query,
                    name=tool["name"],
                    description=tool["description"]
                ) for tool in tools
            ]
            agent = ReActAgent.from_tools(
                agent_tools,
                llm=await ctx.get("llm"),
                verbose=True
            )

            # 使用 asyncio.wait_for 替代 asyncio.timeout
            async def execute_query():
                return await asyncio.wait_for(
                    asyncio.to_thread(agent.chat, ev.question),
                    timeout=180  # 3分钟超时
                )

            # 实现重试逻辑
            max_retries = 3
            retry_delay = 2
            last_exception = None

            for attempt in range(max_retries):
                try:
                    response = await execute_query()
                    
                    ctx.write_event_to_stream(ProgressEvent(
                        status="completed",
                        message=f"Answered: {ev.question}"
                    ))
                    
                    return AnswerEvent(question=ev.question, answer=str(response))

                except asyncio.TimeoutError as e:
                    last_exception = e
                    if attempt < max_retries - 1:
                        wait_time = retry_delay * (2 ** attempt)
                        ctx.write_event_to_stream(ProgressEvent(
                            status="retrying",
                            message=f"Timeout on attempt {attempt + 1}, retrying in {wait_time}s"
                        ))
                        await asyncio.sleep(wait_time)
                    continue
                except Exception as e:
                    last_exception = e
                    if attempt < max_retries - 1:
                        wait_time = retry_delay * (2 ** attempt)
                        ctx.write_event_to_stream(ProgressEvent(
                            status="retrying",
                            message=f"Error on attempt {attempt + 1}, retrying in {wait_time}s: {str(e)}"
                        ))
                        await asyncio.sleep(wait_time)
                    continue

            # 如果所有重试都失败了
            error_message = f"All attempts failed: {str(last_exception)}"
            ctx.write_event_to_stream(ProgressEvent(
                status="error",
                message=error_message
            ))
            raise RuntimeError(error_message)

        except Exception as e:
            ctx.write_event_to_stream(ProgressEvent(
                status="error",
                message=str(e)
            ))
            raise

    @step
    async def combine_answers(self, ctx: Context, ev: AnswerEvent) -> StopEvent:
        ready = ctx.collect_events(ev, [AnswerEvent] * await ctx.get("sub_question_count"))
        if ready is None:
            return None

        try:
            ctx.write_event_to_stream(ProgressEvent(
                status="combining",
                message="Starting to combine answers"
            ))

            # 收集和格式化答案
            formatted_answers = []
            for idx, event in enumerate(ready, 1):
                question = str(event.question) if hasattr(event, 'question') else ''
                answer = str(event.answer) if hasattr(event, 'answer') else ''
                formatted_answers.append(f"Q{idx}: {question}\nA{idx}: {answer.strip()}")

            # 合并答案
            combined_answers = "\n\n".join(formatted_answers)
            original_query = str(await ctx.get('original_query'))
            
            final_prompt = (
                f"Original question: {original_query}\n\n"
                f"Based on these answers:\n\n{combined_answers}\n\n"
                "Provide a comprehensive but concise answer to the original question."
            )

            try:
                llm = await ctx.get("llm")
                final_response = await asyncio.wait_for(
                    asyncio.to_thread(llm.complete, final_prompt),
                    timeout=60
                )
                
                # 确保返回的是简单的字符串
                result = str(final_response).strip()
                
                ctx.write_event_to_stream(ProgressEvent(
                    status="completed",
                    message="Successfully generated final response"
                ))
                
                return StopEvent(result=result)

            except Exception as e:
                error_msg = f"Failed to generate final response: {str(e)}"
                ctx.write_event_to_stream(ProgressEvent(
                    status="error",
                    message=error_msg
                ))
                raise RuntimeError(error_msg)

        except Exception as e:
            error_msg = f"Error in combine_answers: {str(e)}"
            ctx.write_event_to_stream(ProgressEvent(
                status="error",
                message=error_msg
            ))
            raise


def init_openai_settings():
    """初始化OpenAI设置"""
    required_vars = ["OPENAI_API_BASE", "OPENAI_API_KEY", "OPENAI_MODEL", "OPENAI_EMBEDDING_MODEL"]
    
    if missing_vars := [var for var in required_vars if not os.getenv(var)]:
        raise ValueError(f"Missing required environment variables: {', '.join(missing_vars)}")

    Settings.llm = OpenAI(
        api_base=os.getenv("OPENAI_API_BASE"),
        api_key=os.getenv("OPENAI_API_KEY"),
        model=os.getenv("OPENAI_MODEL")
    )
    
    Settings.embed_model = OpenAIEmbedding(
        api_base=os.getenv("OPENAI_API_BASE"),
        api_key=os.getenv("OPENAI_API_KEY"),
        model=os.getenv("OPENAI_EMBEDDING_MODEL")
    )
    
def load_or_create_index(directory_path: str, persist_dir: str) -> VectorStoreIndex:
    """Load existing index or create a new one if it doesn't exist."""
    if not os.path.exists(directory_path):
        raise FileNotFoundError(f"Directory path does not exist: {directory_path}")

    try:
        if os.path.exists(persist_dir):
            print("Loading existing index...")
            storage_context = StorageContext.from_defaults(persist_dir=persist_dir)
            return load_index_from_storage(storage_context)
        
        print("Creating new index...")
        documents = SimpleDirectoryReader(directory_path).load_data()
        index = VectorStoreIndex.from_documents(documents)
        index.storage_context.persist(persist_dir=persist_dir)
        return index
        
    except Exception as e:
        raise RuntimeError(f"Failed to load or create index: {str(e)}")

def create_query_engine():
    """创建查询引擎实例"""
    init_openai_settings()
    
    current_dir = os.path.dirname(os.path.abspath(__file__))
    vector_dir = os.path.join(current_dir, "data", "vector_dir", "paul_graham")
    persist_dir = os.path.join(current_dir, "data", "vector_persist", "paul_graham", "vector_index")
    
    index = load_or_create_index(vector_dir, persist_dir)
    return index.as_query_engine()


def build_sub_question_query_engine_workflow() -> SubQuestionQueryEngine:
    query_engine = create_query_engine()
    return SubQuestionQueryEngine(
        query_engine=query_engine, 
        timeout=300.0, 
        verbose=True
    )

Originally created by @ifreeman6 on GitHub (Nov 27, 2024). When I run `python -m llama_deploy.apiserver` and deploy my workflow, I request the API from front-end `[/deployments/{deployment_name}/tasks/{task_id}/events]`, occurs the following errors in back-end: ```sh ERROR:llama_deploy.services.workflow - Encountered error in task 24576e5e-edff-49cc-b3bf-55008826d818! keys must be str, int, float, bool or None, not ModelMetaclass ``` I try different methods, still not works... help me. Thanks very much! This is my workflow ```python import json from llama_index.core.workflow import ( step, Context, Workflow, Event, StartEvent, StopEvent, ) from llama_index.core.agent import ReActAgent from llama_index.core.settings import Settings from llama_index.llms.openai import OpenAI from llama_index.embeddings.openai import OpenAIEmbedding import os from llama_index.core import SimpleDirectoryReader, VectorStoreIndex, load_index_from_storage, StorageContext from llama_index.core.base.base_query_engine import BaseQueryEngine from llama_index.core.tools import FunctionTool from datetime import datetime from dotenv import load_dotenv import asyncio load_dotenv() class QueryEvent(Event): question: str class AnswerEvent(Event): question: str answer: str class ProgressEvent(Event): def __init__(self, status: str, message: str, timestamp: str = None): super().__init__() self.metadata = { "progress": { "status": status, "message": message, "timestamp": timestamp or datetime.now().isoformat() } } class SubQuestionQueryEngine(Workflow): def __init__(self, query_engine: BaseQueryEngine, *args, **kwargs): super().__init__(*args, **kwargs) self.query_engine = query_engine self.llm = OpenAI( api_base=os.getenv("OPENAI_API_BASE"), api_key=os.getenv("OPENAI_API_KEY"), model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"), timeout=120, max_retries=3 ) self.default_tools = [ { "name": "search", "description": "Search through the document database to find relevant information" } ] @step async def query(self, ctx: Context, ev: StartEvent) -> QueryEvent: try: if hasattr(ev, "query"): await ctx.set("original_query", ev.query) print(f"Query is {await ctx.get('original_query')}") await ctx.set("llm", self.llm) tools = self.default_tools if hasattr(ev, "tools"): tools = ev.tools await ctx.set("tools", tools) # 发送初始进度事件 ctx.write_event_to_stream(ProgressEvent( status="starting", message="Starting query analysis" )) # 修改这里的工具描述获��方式 tool_descriptions = [tool["description"] for tool in tools] response = (await ctx.get("llm")).complete( f""" Given a user question, and a list of tools, output a list of relevant sub-questions (not more than 3), such that the answers to all the sub-questions put together will answer the question. Respond in pure JSON without any markdown, like this: {{ "sub_questions": [ "What is the population of San Francisco?", "What is the budget of San Francisco?", "What is the GDP of San Francisco?" ] }} Here is the user question: {await ctx.get('original_query')} Available tools: {tool_descriptions} """ ) print(f"Sub-questions are {response}") response_obj = json.loads(str(response)) sub_questions = response_obj["sub_questions"] await ctx.set("sub_question_count", len(sub_questions)) # 发送进度事件 ctx.write_event_to_stream(ProgressEvent( status="running", message=f"Generated {len(sub_questions)} sub-questions" )) # 逐个发送子问题 for question in sub_questions: ctx.send_event(QueryEvent(question=question)) return None except Exception as e: ctx.write_event_to_stream(ProgressEvent( status="error", message=str(e) )) raise @step async def sub_question(self, ctx: Context, ev: QueryEvent) -> AnswerEvent: try: ctx.write_event_to_stream(ProgressEvent( status="processing", message=f"Processing sub-question: {ev.question}" )) print(f"Sub-question is {ev.question}") tools = await ctx.get("tools") agent_tools = [ FunctionTool.from_defaults( fn=self.query_engine.query, name=tool["name"], description=tool["description"] ) for tool in tools ] agent = ReActAgent.from_tools( agent_tools, llm=await ctx.get("llm"), verbose=True ) # 使用 asyncio.wait_for 替代 asyncio.timeout async def execute_query(): return await asyncio.wait_for( asyncio.to_thread(agent.chat, ev.question), timeout=180 # 3分钟超时 ) # 实现重试逻辑 max_retries = 3 retry_delay = 2 last_exception = None for attempt in range(max_retries): try: response = await execute_query() ctx.write_event_to_stream(ProgressEvent( status="completed", message=f"Answered: {ev.question}" )) return AnswerEvent(question=ev.question, answer=str(response)) except asyncio.TimeoutError as e: last_exception = e if attempt < max_retries - 1: wait_time = retry_delay * (2 ** attempt) ctx.write_event_to_stream(ProgressEvent( status="retrying", message=f"Timeout on attempt {attempt + 1}, retrying in {wait_time}s" )) await asyncio.sleep(wait_time) continue except Exception as e: last_exception = e if attempt < max_retries - 1: wait_time = retry_delay * (2 ** attempt) ctx.write_event_to_stream(ProgressEvent( status="retrying", message=f"Error on attempt {attempt + 1}, retrying in {wait_time}s: {str(e)}" )) await asyncio.sleep(wait_time) continue # 如果所有重试都失败了 error_message = f"All attempts failed: {str(last_exception)}" ctx.write_event_to_stream(ProgressEvent( status="error", message=error_message )) raise RuntimeError(error_message) except Exception as e: ctx.write_event_to_stream(ProgressEvent( status="error", message=str(e) )) raise @step async def combine_answers(self, ctx: Context, ev: AnswerEvent) -> StopEvent: ready = ctx.collect_events(ev, [AnswerEvent] * await ctx.get("sub_question_count")) if ready is None: return None try: ctx.write_event_to_stream(ProgressEvent( status="combining", message="Starting to combine answers" )) # 收集和格式化答案 formatted_answers = [] for idx, event in enumerate(ready, 1): question = str(event.question) if hasattr(event, 'question') else '' answer = str(event.answer) if hasattr(event, 'answer') else '' formatted_answers.append(f"Q{idx}: {question}\nA{idx}: {answer.strip()}") # 合并答案 combined_answers = "\n\n".join(formatted_answers) original_query = str(await ctx.get('original_query')) final_prompt = ( f"Original question: {original_query}\n\n" f"Based on these answers:\n\n{combined_answers}\n\n" "Provide a comprehensive but concise answer to the original question." ) try: llm = await ctx.get("llm") final_response = await asyncio.wait_for( asyncio.to_thread(llm.complete, final_prompt), timeout=60 ) # 确保返回的是简单的字符串 result = str(final_response).strip() ctx.write_event_to_stream(ProgressEvent( status="completed", message="Successfully generated final response" )) return StopEvent(result=result) except Exception as e: error_msg = f"Failed to generate final response: {str(e)}" ctx.write_event_to_stream(ProgressEvent( status="error", message=error_msg )) raise RuntimeError(error_msg) except Exception as e: error_msg = f"Error in combine_answers: {str(e)}" ctx.write_event_to_stream(ProgressEvent( status="error", message=error_msg )) raise def init_openai_settings(): """初始化OpenAI设置""" required_vars = ["OPENAI_API_BASE", "OPENAI_API_KEY", "OPENAI_MODEL", "OPENAI_EMBEDDING_MODEL"] if missing_vars := [var for var in required_vars if not os.getenv(var)]: raise ValueError(f"Missing required environment variables: {', '.join(missing_vars)}") Settings.llm = OpenAI( api_base=os.getenv("OPENAI_API_BASE"), api_key=os.getenv("OPENAI_API_KEY"), model=os.getenv("OPENAI_MODEL") ) Settings.embed_model = OpenAIEmbedding( api_base=os.getenv("OPENAI_API_BASE"), api_key=os.getenv("OPENAI_API_KEY"), model=os.getenv("OPENAI_EMBEDDING_MODEL") ) def load_or_create_index(directory_path: str, persist_dir: str) -> VectorStoreIndex: """Load existing index or create a new one if it doesn't exist.""" if not os.path.exists(directory_path): raise FileNotFoundError(f"Directory path does not exist: {directory_path}") try: if os.path.exists(persist_dir): print("Loading existing index...") storage_context = StorageContext.from_defaults(persist_dir=persist_dir) return load_index_from_storage(storage_context) print("Creating new index...") documents = SimpleDirectoryReader(directory_path).load_data() index = VectorStoreIndex.from_documents(documents) index.storage_context.persist(persist_dir=persist_dir) return index except Exception as e: raise RuntimeError(f"Failed to load or create index: {str(e)}") def create_query_engine(): """创建查询引擎实例""" init_openai_settings() current_dir = os.path.dirname(os.path.abspath(__file__)) vector_dir = os.path.join(current_dir, "data", "vector_dir", "paul_graham") persist_dir = os.path.join(current_dir, "data", "vector_persist", "paul_graham", "vector_index") index = load_or_create_index(vector_dir, persist_dir) return index.as_query_engine() def build_sub_question_query_engine_workflow() -> SubQuestionQueryEngine: query_engine = create_query_engine() return SubQuestionQueryEngine( query_engine=query_engine, timeout=300.0, verbose=True ) ```

yindo closed this issue 2026-02-16 01:16:01 -05:00

yindo commented 2026-02-16 01:16:02 -05:00

Author

Owner

Copy Link

Copy Source

@ifreeman6 commented on GitHub (Nov 27, 2024):

Always the last step occus error, as the pic shows:

@ifreeman6 commented on GitHub (Nov 27, 2024): Always the last step occus error, as the pic shows:  

yindo commented 2026-02-16 01:16:02 -05:00

Author

Owner

Copy Link

Copy Source

@logan-markewich commented on GitHub (Nov 27, 2024):

@ifreeman6

My prime suspect here is this:

class ProgressEvent(Event):
    def __init__(self, status: str, message: str, timestamp: str = None):
        super().__init__()
        self.metadata = {
            "progress": {
                "status": status,
                "message": message,
                "timestamp": timestamp or datetime.now().isoformat()
            }
        }

Try changing this to

from pydantic import Field

class ProgressEvent(Event):
    status: str
    message: str
    timestamp: str = Field(default_factory=lambda: datetime.now().isoformat())

And then modify your usage of this class

ProgressEvent(status="status", message="message", timestamp=timestamp)

@logan-markewich commented on GitHub (Nov 27, 2024): @ifreeman6 My prime suspect here is this: ```python class ProgressEvent(Event): def __init__(self, status: str, message: str, timestamp: str = None): super().__init__() self.metadata = { "progress": { "status": status, "message": message, "timestamp": timestamp or datetime.now().isoformat() } } ``` Try changing this to ```python from pydantic import Field class ProgressEvent(Event): status: str message: str timestamp: str = Field(default_factory=lambda: datetime.now().isoformat()) ``` And then modify your usage of this class ```python ProgressEvent(status="status", message="message", timestamp=timestamp) ```

yindo commented 2026-02-16 01:16:02 -05:00

Author

Owner

Copy Link

Copy Source

@ifreeman6 commented on GitHub (Nov 27, 2024):

@ifreeman6

My prime suspect here is this:

class ProgressEvent(Event):
    def __init__(self, status: str, message: str, timestamp: str = None):
        super().__init__()
        self.metadata = {
            "progress": {
                "status": status,
                "message": message,
                "timestamp": timestamp or datetime.now().isoformat()
            }
        }

Try changing this to

from pydantic import Field

class ProgressEvent(Event):
    status: str
    message: str
    timestamp: str = Field(default_factory=lambda: datetime.now().isoformat())

And then modify your usage of this class

ProgressEvent(status="status", message="message", timestamp=timestamp)

Thanks your reply, but this change still doesn't work...

@ifreeman6 commented on GitHub (Nov 27, 2024): > @ifreeman6 > > My prime suspect here is this: > > ```python > class ProgressEvent(Event): > def __init__(self, status: str, message: str, timestamp: str = None): > super().__init__() > self.metadata = { > "progress": { > "status": status, > "message": message, > "timestamp": timestamp or datetime.now().isoformat() > } > } > ``` > > Try changing this to > > ```python > from pydantic import Field > > class ProgressEvent(Event): > status: str > message: str > timestamp: str = Field(default_factory=lambda: datetime.now().isoformat()) > ``` > > And then modify your usage of this class > > ```python > ProgressEvent(status="status", message="message", timestamp=timestamp) > ``` Thanks your reply, but this change still doesn't work...

yindo commented 2026-02-16 01:16:02 -05:00

Author

Owner

Copy Link

Copy Source

@ifreeman6 commented on GitHub (Nov 27, 2024):

Although I managed to work around this issue with an indirect solution in the frontend, I suspect the underlying problem might be a system-level bug that requires further investigation.

This is my temporary front-end workaround solution:

// Break out of the stream reading when we get the result (Stop the events stream API when get the final result)
reader.cancel();
break;

The back-end print log as follows:

ERROR:llama_deploy.services.workflow - Encountered error in task 507ab1b0-552b-439b-846e-c6ae853fe225! keys must be str, int, float, bool or None, not ModelMetaclass

ERROR:llama_deploy.services.workflow - Encountered error in task 507ab1b0-552b-439b-846e-c6ae853fe225! keys must be str, int, float, bool or None, not ModelMetaclass
INFO:     127.0.0.1:58055 - "POST /process_message HTTP/1.1" 200 OK
INFO:llama_deploy.message_queues.simple - Successfully published message 'control_plane' to consumer.
INFO:     127.0.0.1:58057 - "POST /process_message HTTP/1.1" 200 OK
INFO:llama_deploy.message_queues.simple - Successfully published message 'control_plane' to consumer.
INFO:     127.0.0.1:57972 - "OPTIONS /deployments/MyDeployment/sessions/delete?session_id=bb921760-c10b-4525-b9ab-978d45d0b357 HTTP/1.1" 200 OK
INFO:     127.0.0.1:58059 - "POST /sessions/bb921760-c10b-4525-b9ab-978d45d0b357/delete HTTP/1.1" 200 OK
INFO:     127.0.0.1:57972 - "POST /deployments/MyDeployment/sessions/delete?session_id=bb921760-c10b-4525-b9ab-978d45d0b357 HTTP/1.1" 200 OK
ERROR:llama_deploy.control_plane.server - Error in event stream for session bb921760-c10b-4525-b9ab-978d45d0b357, task 507ab1b0-552b-439b-846e-c6ae853fe225: 404: Session not found

@ifreeman6 commented on GitHub (Nov 27, 2024): Although I managed to work around this issue with an indirect solution in the frontend, I suspect the underlying problem might be a system-level bug that requires further investigation. This is my temporary front-end workaround solution: ```js // Break out of the stream reading when we get the result (Stop the events stream API when get the final result) reader.cancel(); break; ``` The back-end print log as follows: ERROR:llama_deploy.services.workflow - Encountered error in task 507ab1b0-552b-439b-846e-c6ae853fe225! keys must be str, int, float, bool or None, not ModelMetaclass ```shell ERROR:llama_deploy.services.workflow - Encountered error in task 507ab1b0-552b-439b-846e-c6ae853fe225! keys must be str, int, float, bool or None, not ModelMetaclass INFO: 127.0.0.1:58055 - "POST /process_message HTTP/1.1" 200 OK INFO:llama_deploy.message_queues.simple - Successfully published message 'control_plane' to consumer. INFO: 127.0.0.1:58057 - "POST /process_message HTTP/1.1" 200 OK INFO:llama_deploy.message_queues.simple - Successfully published message 'control_plane' to consumer. INFO: 127.0.0.1:57972 - "OPTIONS /deployments/MyDeployment/sessions/delete?session_id=bb921760-c10b-4525-b9ab-978d45d0b357 HTTP/1.1" 200 OK INFO: 127.0.0.1:58059 - "POST /sessions/bb921760-c10b-4525-b9ab-978d45d0b357/delete HTTP/1.1" 200 OK INFO: 127.0.0.1:57972 - "POST /deployments/MyDeployment/sessions/delete?session_id=bb921760-c10b-4525-b9ab-978d45d0b357 HTTP/1.1" 200 OK ERROR:llama_deploy.control_plane.server - Error in event stream for session bb921760-c10b-4525-b9ab-978d45d0b357, task 507ab1b0-552b-439b-846e-c6ae853fe225: 404: Session not found ```

yindo commented 2026-02-16 01:16:02 -05:00

Author

Owner

Copy Link

Copy Source

@apost71 commented on GitHub (Dec 13, 2024):

@ifreeman6 I encountered this same issue, it seems to be stemming from this line in the Context class. If the key is a pydantic model this dictionary is not serializable here

@apost71 commented on GitHub (Dec 13, 2024): @ifreeman6 I encountered this same issue, it seems to be stemming from [this](https://github.com/run-llama/llama_index/blob/095d410249f6bd8e571275993b418af688ca2daf/llama-index-core/llama_index/core/workflow/context.py#L111) line in the `Context` class. If the key is a pydantic model this dictionary is not serializable [here](https://github.com/run-llama/llama_index/blob/095d410249f6bd8e571275993b418af688ca2daf/llama-index-core/llama_index/core/workflow/context.py#L111)

yindo referenced this issue 2026-02-16 01:16:46 -05:00

[PR #148] [MERGED] Add KafkaMessageQueue #293

Sign in to join this conversation.

No Branch/Tag Specified

Branches Tags

main

dependabot/npm_and_yarn/templates/basic/ui/qs-6.14.2

dependabot/npm_and_yarn/templates/basic/ui/next-16.1.5

dependabot/npm_and_yarn/examples/quick_start/ui/next-16.1.5

dependabot/uv/python-multipart-0.0.22

dependabot/npm_and_yarn/templates/basic/ui/tar-7.5.6

dependabot/npm_and_yarn/templates/basic/ui/modelcontextprotocol/sdk-1.24.0

massi/mcp

adrian/test-autodeploy

adrian/mcp

ms/add-llama-index-server

event_model

v0.8.x

massi/helm

v0.9.2

v0.9.1

v0.9.0

v0.8.1

v0.8.0

v0.7.2

v0.7.1

v0.7.0

v0.6.1

v0.6.0

v0.5.1

v0.5.0

v0.4.4

v0.4.3

v0.4.2

v0.4.1

v0.4.0

v0.3.6

v0.3.5

v0.3.4

v0.3.3

v0.3.2

v0.3.1

v0.3.0

v0.2.4

v0.2.3

v0.2.1

v0.2.0

v0.1.3

v0.1.2

v0.1.1

v0.1.0

v0.0.14

v0.0.13

v0.0.12

v0.0.11

v0.0.10

v0.0.9

v0.0.8

v0.0.7

v0.0.6

v0.0.5

v0.0.4

v0.0.3.post1

v0.0.3

v0.0.2

v0.0.1

Labels

Clear labels

P0

P1

P2

bug

documentation

documentation

enhancement

example

good first issue

invalid

pull-request

Mirrored from GitHub Pull Request

question

topic:CLI

topic:SDK

topic:SDK

topic:SDK

topic:apiserver

topic:control_plane

topic:message_queues

wontfix

No Label

Milestone

No items

No Milestone

Projects

Clear projects

No project

Assignees

Clear assignees

yindo

No Assignees

1 Participants

Notifications

Subscribe

Due Date

No due date set.

Dependencies

No dependencies set.

Reference: run-llama/llama_deploy#148