I am encountering a validation error when trying to configure short-term and long-term memory in CrewAI using custom embeddings. Even after disabling long-term memory (long_term_memory=None), the error persists. The issue seems to be related to the TaskEvaluation model validation.
crew = Crew(
agents=[summarizer_agent, rag_chat_agent, general_chat_agent],
tasks=[chat_task],
process=Process.hierarchical,
verbose=True,
max_iter=3,
manager_agent=manager_agent,
memory=True,
embedder={
“provider”: “custom”,
“config”: {
“embedder”: SentenceTransformerEmbedder(EMBEDDING_MODEL_PATH)
}
},
short_term_memory=ShortTermMemory(
storage=RAGStorage(
embedder_config={
“provider”: “custom”,
“config”: {
“embedder”: SentenceTransformerEmbedder(EMBEDDING_MODEL_PATH)
}
},
type=“short_term”,
path=SHORT_TERM_PATH
)
),
long_term_memory=LongTermMemory(
storage=LTMSQLiteStorage(
db_path=LONG_TERM_PATH
)
),
)
Error Message:
I receive the following error when running the code:
Failed to add to long term memory: Failed to convert text into a Pydantic model due to the following error: 3 validation errors for TaskEvaluation
suggestions
Field required [type=missing, input_value={‘task_description’: ‘{“d… of France is Paris.”}’}, input_type=dict]
For further information visit Redirecting...
quality
Field required [type=missing, input_value={‘task_description’: ‘{“d… of France is Paris.”}’}, input_type=dict]
What I’ve Tried:
- Tried setting
long_term_memory=None —> Same error persists. so pls any one faced same issue give the soliton my crew ai everying recent version crewai 0.102
Hi, Rajesh! I suggest that you post here the parameters that you are using for your chat_task. It’s also important to present the definition of the BaseModel that you’re passing to the output_pydantic parameter of that task.
def create_autonomous_task(user_query: str):
has_document = “collection” in st.session_state
return Task(
description=f"""Analyze and process the user query:
User Query: {user_query}
Document Status: {‘Uploaded’ if has_document else ‘Not Uploaded’}
{user_query}
Available agent specializations:
1. Document Summarization Expert - .docx/.pdf summaries
2. Document Analysis Specialist - Document content questions
3. Conversational Assistant - General chat
Routing Rules:
1. If query contains "summarize" and document exists -> Summarizer
2. If document exists and question references content -> Document Analyst
3. All other cases -> General Chat
""",
expected_output="Immediate, final response prefixed with 'Final Answer:'",
context=[{
"has_document": has_document,
"query": user_query,
"description": user_query,
"expected_output": "Relevant response matching query intent"
}],
config={
"allow_delegation": True,
"stop_on_success": True
}
)
chat_task = create_autonomous_task(user_input)
I’m not using output_pydantic directly — could that be the issue? Should I define a specific BaseModel for the output_pydantic field to satisfy Pydantic’s validation?
If I set memory=False, I immediately get a response. However, when I set memory=True with the following short_term_memory configuration:
short_term_memory=ShortTermMemory(
storage=RAGStorage(
embedder_config={
“provider”: “custom”,
“config”: {
“embedder”: SentenceTransformerEmbedder(EMBEDDING_MODEL_PATH)
}
},
type=“short_term”,
path=SHORT_TERM_PATH
)
)
I still get the following error, even if long_term_memory=None:
Failed to add to long term memory: Failed to convert text into a Pydantic model due to the following error: 3 validation errors for TaskEvaluation
suggestions
Field required [type=missing, input_value={‘task_description’: ‘{“d… of France is Paris.”}’}, input_type=dict]
For further information visit Redirecting...
quality
Field required [type=missing, input_value={‘task_description’: ‘{“d… of France is Paris.”}’}, input_type=dict]
For further information visit Redirecting...
entities
For further information visit https://errors.pydantic.dev/2.10/v/missing
To fix this, I tried adding long_term_memory with the following configuration:
long_term_memory=LongTermMemory(
storage=LTMSQLiteStorage(
db_path=LONG_TERM_PATH
)
)
However, the issue persists. I am using a custom embedding model with the following class:
class SentenceTransformerEmbedder(EmbeddingFunction):
def init(self, model_path: str):
self.model = SentenceTransformer(model_path)
super().init()
def __call__(self, input: Documents) -> EMD:
return self.model.encode(input).tolist()
def encode_documents(self, documents: List[str]) -> List[List[float]]:
return self.model.encode(documents).tolist()
def encode_queries(self, queries: List[str]) -> List[List[float]]:
return self.model.encode(queries).tolist()
Wow, lots of things going on here. I swear I did my best to understand and contribute. After reformatting your code and evaluating the scenario you presented, these are my considerations:
- The
context parameter does indeed exist in a Task. However, let’s look at what the documentation says about this parameter: “Other tasks whose outputs will be used as context for this task.” Basically, you’ll use it when your current task depends on the output of a previous task (it can be omitted in the case of Process.sequential). I suggest you review the documentation on tasks, especially the “Task Attributes” section.
- The
config parameter does indeed exist in a Task. However, I don’t see the point of using it in your use case, since the information you were trying to pass can simply be included in the description of your task (see how I included it in the reformatted version that I present below).
- It wasn’t clear if you’re trying to implement a hierarchical
Process. If that’s the case, I suggest you review the documentation on processes, especially on “Hierarchical Process”. I also suggest reading this thread on the subject.
- As for the use of
Memory, I suggest you review the documentation on the subject, especially the sections that exemplify customizations.
- Finally, I suggest that you first successfully execute a super simplified version of the solution you’re proposing, trying to ensure that each small part of your solution is working as you want, and then increase the complexity to meet your effective use case.
Here’s the version of your code that I reformatted as I understood it:
def create_autonomous_task(user_query: str):
has_document = "collection" in st.session_state
document_status = "Uploaded" if has_document else "Not Uploaded"
return Task(
description=(f"""
Analyze and process the user query:
User Query: {user_query}
Available agent specializations:
1. Document Summarization Expert - .docx/.pdf summaries
2. Document Analysis Specialist - Document content questions
3. Conversational Assistant - General chat
Routing Rules:
1. If query contains "summarize" and document exists -> Summarizer
2. If document exists and question references content -> Document Analyst
3. All other cases -> General Chat
Context:
- Document Status: {document_status}
- Document Exists: {has_document}
"""),
expected_output="Immediate, final response prefixed with 'Final Answer:'"
)
chat_task = create_autonomous_task(user_input)