History scope

aworld/agents/llm_agent.py

@@ -252,6 +252,45 @@ def messages_transform(self,
         return sync_exec(self.async_messages_transform, image_urls=image_urls, observation=observation,
                          message=message, **kwargs)
 
+    def _is_amni_context(self, context: Context):
+        from aworld.core.context.amni import AmniContext
+        return isinstance(context, AmniContext)
+
+    def _build_memory_filters(self, context: Context, additional_filters: Dict[str, Any] = None) -> Dict[str, Any]:
+        filters = {
+            "agent_id": self.id()
+        }
+
+        # Decide which filter to add based on history_scope
+        agent_memory_config = self.memory_config
+        if self._is_amni_context(context):
+            agent_context_config = context.get_config().get_agent_context_config(self.id())
+            agent_memory_config = agent_context_config.to_memory_config()
+
+        query_scope = agent_memory_config.history_scope if agent_memory_config and agent_memory_config.history_scope else "task"
+        task = context.get_task()
+
+        if query_scope == "user":
+            # Pass user_id when query_scope is user
+            if hasattr(context, 'user_id') and context.user_id:
+                filters["user_id"] = context.user_id
+            elif hasattr(task, 'user_id') and task.user_id:
+                filters["user_id"] = task.user_id
+        elif query_scope == "session":
+            # Pass session_id when query_scope is session
+            if task and task.session_id:
+                filters["session_id"] = task.session_id
+        else:  # query_scope == "task" or default
+            # Pass task_id when query_scope is task
+            if task and task.id:
+                filters["task_id"] = task.id
+
+        # Add additional filter conditions
+        if additional_filters:
+            filters.update(additional_filters)
+
+        return filters
+
     def _clean_redundant_tool_call_messages(self, histories: List[MemoryItem]) -> None:
         try:
             for i in range(len(histories) - 1, -1, -1):
@@ -269,14 +308,8 @@ def postprocess_terminate_loop(self, message: Message):
         logger.info(f"Agent {self.id()} postprocess_terminate_loop: {self.loop_step}")
         super().postprocess_terminate_loop(message)
         try:
-            session_id = message.context.get_task().session_id
-            task_id = message.context.get_task().id
-            histories = self.memory.get_all(filters={
-                "agent_id": self.id(),
-                "session_id": session_id,
-                "task_id": task_id,
-                "memory_type": "message"
-            })
+            filters = self._build_memory_filters(message.context, additional_filters={"memory_type": "message"})
+            histories = self.memory.get_all(filters=filters)
             self._clean_redundant_tool_call_messages(histories)
         except Exception:
             logger.error(f"Agent {self.id()} postprocess_terminate_loop error: {traceback.format_exc()}")
@@ -304,14 +337,8 @@ async def async_messages_transform(self,
         if self.system_prompt:
             await self._add_message_to_memory(context=message.context, payload=content, message_type=MemoryType.SYSTEM)
 
-        session_id = message.context.get_task().session_id
-        task_id = message.context.get_task().id
-        histories = self.memory.get_all(filters={
-            "agent_id": self.id(),
-            "session_id": session_id,
-            "task_id": task_id,
-            "memory_type": "message"
-        })
+        filters = self._build_memory_filters(message.context, additional_filters={"memory_type": "message"})
+        histories = self.memory.get_all(filters=filters)
 
         # append observation to memory
         tool_result_added = False
@@ -333,11 +360,12 @@ async def async_messages_transform(self,
                                               context=message.context)
 
         # from memory get last n messages
-        histories = self.memory.get_last_n(self.memory_config.history_rounds, filters={
-            "agent_id": self.id(),
-            "session_id": session_id,
-            "task_id": task_id
-        }, agent_memory_config=self.memory_config)
+        filters = self._build_memory_filters(message.context)
+        agent_memory_config = self.memory_config
+        if self._is_amni_context(message.context):
+            agent_context_config = message.context.get_config().get_agent_context_config(self.id())
+            agent_memory_config = agent_context_config.to_memory_config()
+        histories = self.memory.get_last_n(agent_memory_config.history_rounds, filters=filters, agent_memory_config=agent_memory_config)
         if histories:
             tool_calls_map = {}
             last_tool_calls = []
@@ -841,12 +869,8 @@ async def _add_tool_result_token_ids_to_context(self, context: Context):
         """Add tool result token ids to context"""
         if context.get_task().conf.get("run_mode") != TaskRunMode.INTERACTIVE:
             return
-        histories = self.memory.get_all(filters={
-            "agent_id": self.id(),
-            "session_id": context.get_task().session_id,
-            "task_id": context.get_task().id,
-            "memory_type": "message"
-        })
+        filters = self._build_memory_filters(context, additional_filters={"memory_type": "message"})
+        histories = self.memory.get_all(filters=filters)
         tool_openai_messages_after_last_assistant = []
         found_assistant = False
         tool_call_ids = []

aworld/config/conf.py

@@ -188,7 +188,8 @@ class AgentMemoryConfig(BaseConfig):
                                 description="rounds of message msg; when the number of messages is greater than the history_rounds, the memory will be trimmed")
     history_write_strategy: HistoryWriteStrategy = Field(default=HistoryWriteStrategy.EVENT_DRIVEN,
                                                          description="History write strategy: event_driven (through message system) or direct (direct call to handler)")
-
+    history_scope: Optional[str] = Field(default="task", description="History initialization scope: user, session, or task")
+
     enable_summary: bool = Field(default=False,
                                  description="enable_summary use llm to create summary short-term memory")
     summary_model: Optional[str] = Field(default=None, description="short-term summary model")

aworld/core/context/amni/config.py

@@ -90,6 +90,7 @@ class AgentContextConfig(BaseConfig):
                                 description="rounds of message msg; when the number of messages is greater than the history_rounds, the memory will be trimmed")
     history_write_strategy: HistoryWriteStrategy = Field(default=HistoryWriteStrategy.EVENT_DRIVEN,
                                                          description="History write strategy: event_driven (through message system) or direct (direct call to handler)")
+    history_scope: Optional[str] = Field(default="task", description="History initialization scope: user, session, or task")
 
     # Context Reduce - Compress
     enable_summary: bool = Field(default=False,
@@ -118,6 +119,7 @@ def to_memory_config(self) -> AgentMemoryConfig:
         return AgentMemoryConfig(
             history_rounds=self.history_rounds,
             history_write_strategy=self.history_write_strategy,
+            history_scope=self.history_scope,
             enable_summary=self.enable_summary,
             summary_rounds=self.summary_rounds,
             summary_context_length=self.summary_context_length,

aworld/core/tool/base.py

@@ -501,6 +501,15 @@ async def post_step(self,
                                 headers={"context": context})
         return result
 
+        # tool hooks
+        try:
+            events = []
+            async for event in run_hooks(context=message.context, hook_point=HookPoint.POST_TOOL_CALL, hook_from=result.caller, payload=step_res):
+                events.append(event)
+        except Exception:
+            logger.debug(traceback.format_exc())
+        return result
+
     async def _exec_tool_callback(self, step_res: Tuple[Observation, float, bool, bool, Dict[str, Any]],
                                   action: List[ActionModel],
                                   message: Message,

aworld/dataset/trajectory_strategy.py

@@ -312,18 +312,18 @@ async def generate_trajectory_for_memory(self, swarm: Swarm, context: Context):
         }, agent_memory_config=swarm.cur_agent[0].memory_config)
 
         # Convert memory items to OpenAI message format
-        result = {}
+        result = []
         for i, item in enumerate(memory_items):
             # Check if item has to_openai_message method
             if hasattr(item, 'to_openai_message'):
                 message = item.to_openai_message()
                 # Add usage to the message if it exists in metadata
                 if hasattr(item, 'metadata') and item.metadata and 'usage' in item.metadata:
                     message['usage'] = item.metadata['usage']
-                result[i] = message
+                result.append(message)
             else:
                 # If item doesn't have to_openai_message, return the item as is
-                result[i] = item
+                result.append(item)
 
         return result
 

aworld/evaluations/scorers/llm_as_judge.py

@@ -60,7 +60,7 @@ def build_judge_prompt(self, index: int, input: EvalDataCase[EvalCaseDataType],
         raise NotImplementedError("build_judge_prompt must be implemented in subclasses")
 
     @abc.abstractmethod
-    def build_judge_data(self, index: int, input: EvalDataCase[EvalCaseDataType], output: dict) -> str:
+    def build_judge_data(self, index: int, input: EvalDataCase[EvalCaseDataType], output: dict) -> [str, dict]:
         """Builds the input for the judge agent task.
 
         Args:
@@ -72,7 +72,7 @@ def build_judge_data(self, index: int, input: EvalDataCase[EvalCaseDataType], ou
             str: The input string for the judge agent task.
 
         Example:
-            [Question]: {input.case_data.get('question', '')}
+            [疑问]: {input.case_data.get('question', '')}
             [Correct_Answer]: {input.case_data.get('answer', '')}
             [Response]: {output.get('answer', '')}
         """
@@ -106,6 +106,8 @@ async def score(self, index: int, input: EvalDataCase[EvalCaseDataType], output:
                             agent_prompt=self.build_judge_prompt(index=index, input=input, output=output))
 
         task_input = self.build_judge_data(index=index, input=input, output=output)
+        if not task_input:
+            return ScorerResult(scorer_name=self.name, metric_results={})
         response = await exec_agent(task_input, agent=score_agent, context=Context())
         metric_results = self.convert_judge_response_to_score(response.answer)
         if metric_results:
@@ -130,4 +132,4 @@ def _build_judge_system_prompt(self) -> str:
         """
         return '''
         You are a judge model that evaluates the quality of the response.
-        '''
+        '''

aworld/evaluations/scorers/metrics.py

@@ -2,4 +2,5 @@ class MetricNames:
     LABEL_DISTRIBUTION = 'label_distribution'
     SUMMARIZE_QUALITY = 'summarize_quality'
     ANSWER_ACCURACY = 'answer_accuracy'
-    PREDICT_TIME_COST_MS = 'predict_time_cost_ms'
+    PREDICT_TIME_COST_MS = 'predict_time_cost_ms'
+    FLIGHT_JUDGE = 'flight_judge'

aworld/output/workspace.py

@@ -495,6 +495,23 @@ def save(self) -> None:
         self.repository.save_index(workspace_data)
         self._rebuild_artifact_id_index()
 
+    def get_raw_file_content_by_artifact_id(self, artifact_id: str) -> str:
+        """
+        Get concatenated content of all artifacts with the same filename.
+
+        Args:
+            artifact_id: artifact_id
+
+        Returns:
+            Raw unescaped concatenated content of all matching artifacts
+        """
+        filename = artifact_id
+        artifact_data = self.repository.retrieve_latest_artifact(artifact_id)
+        if not artifact_data:
+            return ""
+        artifact = Artifact.from_dict(artifact_data)
+        return artifact.content
+
     def get_file_content_by_artifact_id(self, artifact_id: str) -> str:
         """
         Get concatenated content of all artifacts with the same filename.

aworld/runners/handler/memory.py

@@ -1,22 +1,21 @@
 # aworld/runners/handler/output.py
 import copy
-import json
 import time
 import traceback
 from datetime import datetime
-from typing import AsyncGenerator, Any
+from typing import Any
 
 from aworld.agents.llm_agent import Agent
 from aworld.config import ConfigDict
+from aworld.core.common import ActionResult
 from aworld.core.context.base import Context
+from aworld.core.event.base import Message, Constants, MemoryEventMessage, MemoryEventType
+from aworld.logs.util import logger
 from aworld.memory.main import MemoryFactory
 from aworld.memory.models import MemoryToolMessage, MessageMetadata, MemoryHumanMessage, MemorySystemMessage, \
     MemoryAIMessage
 from aworld.runners import HandlerFactory
 from aworld.runners.handler.base import DefaultHandler
-from aworld.core.common import TaskItem, ActionResult
-from aworld.core.event.base import Message, Constants, TopicType, MemoryEventMessage, MemoryEventType
-from aworld.logs.util import logger
 from aworld.runners.hook.hook_factory import HookFactory
 
 
@@ -182,7 +181,7 @@ async def _add_llm_response_to_memory(self, agent: Agent, llm_response, context:
         """Add LLM response to memory"""
         # Get start time from context (if exists)
         start_time = context.context_info.get("llm_call_start_time")
-        
+
         ai_message = MemoryAIMessage(
             content=llm_response.content,
             tool_calls=llm_response.tool_calls,
@@ -197,13 +196,13 @@ async def _add_llm_response_to_memory(self, agent: Agent, llm_response, context:
                 }
             )
         )
-        
+
         # If start time exists in context, update it
         if start_time:
             ai_message.start_time = start_time
         # Record message end time
         ai_message.end_time = None
-        
+
         agent_memory_config = agent.memory_config
         if self._is_amni_context(context):
             agent_memory_config = context.get_config().get_agent_memory_config(agent.id())
@@ -283,10 +282,10 @@ async def _do_add_tool_result_to_memory(self, agent: 'Agent', tool_call_id: str,
         tool_use_summary = None
         if isinstance(tool_result, ActionResult):
             tool_use_summary = tool_result.metadata.get("tool_use_summary")
-        
+
         # Get start time from context (if exists)
         start_time = context.context_info.get(f"tool_call_start_time_{tool_call_id}")
-        
+
         tool_message = MemoryToolMessage(
             content=tool_result.content if hasattr(tool_result, 'content') else tool_result,
             tool_call_id=tool_call_id,
@@ -301,14 +300,14 @@ async def _do_add_tool_result_to_memory(self, agent: 'Agent', tool_call_id: str,
                 ext_info={"tool_name": tool_result.tool_name, "action_name": tool_result.action_name}
             )
         )
-        
+
         # If start time exists in context, update it
         if start_time:
             tool_message.start_time = start_time
-        
+
         # Record message end time
         tool_message.end_time = None
-        
+
         await memory.add(tool_message, agent_memory_config=agent.memory_config)
 
     def _is_amni_context(self, context: Context):
@@ -318,7 +317,7 @@ def _is_amni_context(self, context: Context):
     @staticmethod
     async def handle_memory_message_directly(memory_msg: MemoryEventMessage, context: Context):
         """Handle memory message directly without going through message system
-        
+
         Args:
             memory_msg: Memory event message
             context: Context object
@@ -329,7 +328,7 @@ class SimpleRunner:
                 def __init__(self, task):
                     self.task = task
                     self.start_time = 0
-            
+
             task = context.get_task()
             simple_runner = SimpleRunner(task)
             handler = DefaultMemoryHandler(simple_runner)

aworld/runners/hook/hooks.py

@@ -13,6 +13,7 @@ class HookPoint:
     ERROR = "error"
     PRE_LLM_CALL = "pre_llm_call"
     POST_LLM_CALL = "post_llm_call"
+    POST_TOOL_CALL = "post_tool_call"
     OUTPUT_PROCESS = "output_process"
     PRE_TOOL_CALL = "pre_tool_call"
     POST_TOOL_CALL = "post_tool_call"