chore: Release 1.7.3 (#3846)

# Changelog

## New Features:

- **Session State on Run**: You can now pass `session_state` when
running an agent or team. See [docs](https://docs.agno.com/agents/state)
for more information.
- **GCS Support for PDF Knowledge Base:** Added `GCSPDFKnowledgeBase` to
support PDFs on Google Cloud Storage.

## Bug Fixes:

- **Workflows Async + Storage**: Fixed issues where sessions was not
correctly stored with async workflow executions.
- **Session State Management**: Fixed issues around session state
management in teams and agents. Session state and session name will now
correctly reset and load from storage if sessions are switched.
- **Metadata Support for Document Knowledge Base:** Adds metadata
support for `DocumentKnowledgeBase`
- **Session Metrics with History**: Fixed bug with session metrics on
`Agent` where history is enabled.

Author: dirkbrnd

cookbook/agent_concepts/knowledge/gcs_pdf_kb.py

@@ -5,7 +5,7 @@
 1. Install required libraries: agno, google-cloud-storage, psycopg2-binary (for PostgreSQL vector DB).
 2. Set up your GCS bucket and upload your PDF file.
 3. For public GCS buckets: No authentication needed, just set the bucket and PDF path.
-4. For private GCS buckets: 
+4. For private GCS buckets:
    - Grant the service account Storage Object Viewer access to the bucket via Google Cloud Console
    - Export GOOGLE_APPLICATION_CREDENTIALS with the path to your service account JSON before running the script
 5. Update 'bucket_name' and 'blob_name' in the script to your PDF's location.

cookbook/agent_concepts/knowledge/gcs_pdf_kb_async.py

@@ -5,14 +5,15 @@
 1. Install required libraries: agno, google-cloud-storage, psycopg2-binary (for PostgreSQL vector DB).
 2. Set up your GCS bucket and upload your PDF file.
 3. For public GCS buckets: No authentication needed, just set the bucket and PDF path.
-4. For private GCS buckets: 
+4. For private GCS buckets:
    - Grant the service account Storage Object Viewer access to the bucket via Google Cloud Console
    - Export GOOGLE_APPLICATION_CREDENTIALS with the path to your service account JSON before running the script
 5. Update 'bucket_name' and 'blob_name' in the script to your PDF's location.
 6. Run the script to load the knowledge base and ask questions.
 """
 
 import asyncio
+
 from agno.agent import Agent
 from agno.knowledge.gcs.pdf import GCSPDFKnowledgeBase
 from agno.vectordb.pgvector import PgVector

cookbook/agent_concepts/other/dynamic_instructions.py

@@ -1,7 +1,9 @@
 from agno.agent import Agent
 
+
 def get_instructions(agent: Agent):
-    return f"Make the story about {agent.session_state.get("current_user_id")}."
+    return f"Make the story about {agent.session_state.get('current_user_id')}."
+
 
 agent = Agent(instructions=get_instructions)
 agent.print_response("Write a 2 sentence story", user_id="john.doe")

cookbook/agent_concepts/state/session_state_on_run.py

@@ -0,0 +1,30 @@
+from agno.agent import Agent
+from agno.models.openai import OpenAIChat
+
+agent = Agent(
+    model=OpenAIChat(id="gpt-4o-mini"),
+    add_state_in_messages=True,
+    instructions="Users name is {user_name} and age is {age}",
+)
+
+# Sets the session state for the session with the id "user_1_session_1"
+agent.print_response(
+    "What is my name?",
+    session_id="user_1_session_1",
+    user_id="user_1",
+    session_state={"user_name": "John", "age": 30},
+)
+
+# Will load the session state from the session with the id "user_1_session_1"
+agent.print_response("How old am I?", session_id="user_1_session_1", user_id="user_1")
+
+# Sets the session state for the session with the id "user_2_session_1"
+agent.print_response(
+    "What is my name?",
+    session_id="user_2_session_1",
+    user_id="user_2",
+    session_state={"user_name": "Jane", "age": 25},
+)
+
+# Will load the session state from the session with the id "user_2_session_1"
+agent.print_response("How old am I?", session_id="user_2_session_1", user_id="user_2")

libs/agno/agno/agent/agent.py

@@ -758,6 +758,7 @@ def _run(
         session_id: str,
         user_id: Optional[str] = None,
         response_format: Optional[Union[Dict, Type[BaseModel]]] = None,
+        refresh_session_before_write: Optional[bool] = False,
     ) -> RunResponse:
         """Run the Agent and return the RunResponse.
 
@@ -825,7 +826,7 @@ def _run(
         self._convert_response_to_structured_format(run_response)
 
         # 6. Save session to storage
-        self.write_to_storage(user_id=user_id, session_id=session_id)
+        self.write_to_storage(user_id=user_id, session_id=session_id, refresh_session=refresh_session_before_write)
 
         # 7. Save output to file if save_response_to_file is set
         self.save_run_response_to_file(message=run_messages.user_message, session_id=session_id)
@@ -845,6 +846,7 @@ def _run_stream(
         user_id: Optional[str] = None,
         response_format: Optional[Union[Dict, Type[BaseModel]]] = None,
         stream_intermediate_steps: bool = False,
+        refresh_session_before_write: Optional[bool] = False,
     ) -> Iterator[RunResponseEvent]:
         """Run the Agent and yield the RunResponse.
 
@@ -917,7 +919,7 @@ def _run_stream(
             yield self._handle_event(create_run_response_completed_event(from_run_response=run_response), run_response)
 
         # 7. Save session to storage
-        self.write_to_storage(user_id=user_id, session_id=session_id)
+        self.write_to_storage(user_id=user_id, session_id=session_id, refresh_session=refresh_session_before_write)
 
         # Log Agent Run
         self._log_agent_run(user_id=user_id, session_id=session_id)
@@ -941,6 +943,7 @@ def run(
         messages: Optional[Sequence[Union[Dict, Message]]] = None,
         retries: Optional[int] = None,
         knowledge_filters: Optional[Dict[str, Any]] = None,
+        refresh_session_before_write: Optional[bool] = False,
         **kwargs: Any,
     ) -> RunResponse: ...
 
@@ -961,6 +964,7 @@ def run(
         messages: Optional[Sequence[Union[Dict, Message]]] = None,
         retries: Optional[int] = None,
         knowledge_filters: Optional[Dict[str, Any]] = None,
+        refresh_session_before_write: Optional[bool] = False,
         **kwargs: Any,
     ) -> Iterator[RunResponseEvent]: ...
 
@@ -980,6 +984,7 @@ def run(
         messages: Optional[Sequence[Union[Dict, Message]]] = None,
         retries: Optional[int] = None,
         knowledge_filters: Optional[Dict[str, Any]] = None,
+        refresh_session_before_write: Optional[bool] = False,
         **kwargs: Any,
     ) -> Union[RunResponse, Iterator[RunResponseEvent]]:
         """Run the Agent and return the response."""
@@ -1109,6 +1114,7 @@ def run(
                         session_id=session_id,
                         response_format=response_format,
                         stream_intermediate_steps=stream_intermediate_steps,
+                        refresh_session_before_write=refresh_session_before_write,
                     )
                     return response_iterator
                 else:
@@ -1118,6 +1124,7 @@ def run(
                         user_id=user_id,
                         session_id=session_id,
                         response_format=response_format,
+                        refresh_session_before_write=refresh_session_before_write,
                     )
                     return response
             except ModelProviderError as e:
@@ -1167,6 +1174,7 @@ async def _arun(
         session_id: str,
         user_id: Optional[str] = None,
         response_format: Optional[Union[Dict, Type[BaseModel]]] = None,
+        refresh_session_before_write: Optional[bool] = False,
     ) -> RunResponse:
         """Run the Agent and yield the RunResponse.
 
@@ -1233,7 +1241,7 @@ async def _arun(
         self._convert_response_to_structured_format(run_response)
 
         # 6. Save session to storage
-        self.write_to_storage(user_id=user_id, session_id=session_id)
+        self.write_to_storage(user_id=user_id, session_id=session_id, refresh_session=refresh_session_before_write)
 
         # 7. Save output to file if save_response_to_file is set
         self.save_run_response_to_file(message=run_messages.user_message, session_id=session_id)
@@ -1253,6 +1261,7 @@ async def _arun_stream(
         user_id: Optional[str] = None,
         response_format: Optional[Union[Dict, Type[BaseModel]]] = None,
         stream_intermediate_steps: bool = False,
+        refresh_session_before_write: Optional[bool] = False,
     ) -> AsyncIterator[RunResponseEvent]:
         """Run the Agent and yield the RunResponse.
 
@@ -1328,7 +1337,7 @@ async def _arun_stream(
             yield self._handle_event(create_run_response_completed_event(from_run_response=run_response), run_response)
 
         # 7. Save session to storage
-        self.write_to_storage(user_id=user_id, session_id=session_id)
+        self.write_to_storage(user_id=user_id, session_id=session_id, refresh_session=refresh_session_before_write)
 
         # Log Agent Run
         await self._alog_agent_run(user_id=user_id, session_id=session_id)
@@ -1351,6 +1360,7 @@ async def arun(
         stream_intermediate_steps: Optional[bool] = None,
         retries: Optional[int] = None,
         knowledge_filters: Optional[Dict[str, Any]] = None,
+        refresh_session_before_write: Optional[bool] = False,
         **kwargs: Any,
     ) -> Any:
         """Async Run the Agent and return the response."""
@@ -1361,6 +1371,9 @@ async def arun(
 
         log_debug(f"Session ID: {session_id}", center=True)
 
+        # Initialize the Agent
+        self.initialize_agent()
+
         effective_filters = knowledge_filters
         # When filters are passed manually
         if self.knowledge_filters or knowledge_filters:
@@ -1476,6 +1489,7 @@ async def arun(
                         session_id=session_id,
                         response_format=response_format,
                         stream_intermediate_steps=stream_intermediate_steps,
+                        refresh_session_before_write=refresh_session_before_write,
                     )  # type: ignore[assignment]
                     return response_iterator
                 else:
@@ -1485,6 +1499,7 @@ async def arun(
                         user_id=user_id,
                         session_id=session_id,
                         response_format=response_format,
+                        refresh_session_before_write=refresh_session_before_write,
                     )
                     return response
             except ModelProviderError as e:
@@ -4123,13 +4138,56 @@ def read_from_storage(
                 self.load_agent_session(session=self.agent_session)
         return self.agent_session
 
-    def write_to_storage(self, session_id: str, user_id: Optional[str] = None) -> Optional[AgentSession]:
+    def refresh_from_storage(self, session_id: str) -> None:
+        """Refresh the AgentSession from storage
+
+        Args:
+            session_id: The session_id to refresh from storage.
+        """
+        if not self.storage:
+            return
+
+        agent_session_from_db = self.storage.read(session_id=session_id)
+        if (
+            agent_session_from_db is not None
+            and agent_session_from_db.memory is not None
+            and "runs" in agent_session_from_db.memory  # type: ignore
+        ):
+            if isinstance(self.memory, AgentMemory):
+                return
+            try:
+                if self.memory.runs is None:  # type: ignore
+                    self.memory.runs = {}  # type: ignore
+                if session_id not in self.memory.runs:  # type: ignore
+                    self.memory.runs[session_id] = []  # type: ignore
+                for run in agent_session_from_db.memory["runs"]:  # type: ignore
+                    run_session_id = run["session_id"]
+                    skip = False
+                    for existing_run in self.memory.runs[run_session_id]:  # type: ignore
+                        if existing_run.run_id == run["run_id"]:
+                            skip = True
+                            break
+                    if skip:
+                        continue
+                    if "team_id" in run:
+                        self.memory.runs[run_session_id].append(TeamRunResponse.from_dict(run))  # type: ignore
+                    else:
+                        self.memory.runs[run_session_id].append(RunResponse.from_dict(run))  # type: ignore
+            except Exception as e:
+                log_warning(f"Failed to load runs from memory: {e}")
+
+    def write_to_storage(
+        self, session_id: str, user_id: Optional[str] = None, refresh_session: Optional[bool] = False
+    ) -> Optional[AgentSession]:
         """Save the AgentSession to storage
 
         Returns:
             Optional[AgentSession]: The saved AgentSession or None if not saved.
         """
         if self.storage is not None:
+            if refresh_session:
+                self.refresh_from_storage(session_id=session_id)
+
             self.agent_session = cast(
                 AgentSession,
                 self.storage.upsert(session=self.get_agent_session(session_id=session_id, user_id=user_id)),
@@ -4209,13 +4267,13 @@ def new_session(self) -> None:
         self.session_id = str(uuid4())
         self.load_session(force=True)
 
-    def format_message_with_state_variables(self, msg: Any) -> Any:
+    def format_message_with_state_variables(self, message: Any) -> Any:
         """Format a message with the session state variables."""
         import re
         import string
 
-        if not isinstance(msg, str):
-            return msg
+        if not isinstance(message, str):
+            return message
 
         format_variables = ChainMap(
             self.session_state or {},
@@ -4224,7 +4282,7 @@ def format_message_with_state_variables(self, msg: Any) -> Any:
             self.extra_data or {},
             {"user_id": self.user_id} if self.user_id is not None else {},
         )
-        converted_msg = msg
+        converted_msg = message
         for var_name in format_variables.keys():
             # Only convert standalone {var_name} patterns, not nested ones
             pattern = r"\{" + re.escape(var_name) + r"\}"
@@ -4238,7 +4296,7 @@ def format_message_with_state_variables(self, msg: Any) -> Any:
             return result
         except Exception as e:
             log_warning(f"Template substitution failed: {e}")
-            return msg
+            return message
 
     def get_system_message(self, session_id: str, user_id: Optional[str] = None) -> Optional[Message]:
         """Return the system message for the Agent.
@@ -4264,6 +4322,7 @@ def get_system_message(self, session_id: str, user_id: Optional[str] = None) ->
             # Format the system message with the session state variables
             if self.add_state_in_messages:
                 sys_message_content = self.format_message_with_state_variables(sys_message_content)
+                print("HELLO", sys_message_content)
 
             # Add the JSON output prompt if response_model is provided and the model does not support native structured outputs or JSON schema outputs
             # or if use_json_mode is True
@@ -4280,7 +4339,6 @@ def get_system_message(self, session_id: str, user_id: Optional[str] = None) ->
 
             # type: ignore
             return Message(role=self.system_message_role, content=sys_message_content)
-
         # 2. If create_default_system_message is False, return None.
         if not self.create_default_system_message:
             return None
@@ -4295,6 +4353,7 @@ def get_system_message(self, session_id: str, user_id: Optional[str] = None) ->
             _instructions = self.instructions
             if callable(self.instructions):
                 import inspect
+
                 signature = inspect.signature(self.instructions)
                 if "agent" in signature.parameters:
                     _instructions = self.instructions(agent=self)
@@ -4305,7 +4364,7 @@ def get_system_message(self, session_id: str, user_id: Optional[str] = None) ->
                 instructions.append(_instructions)
             elif isinstance(_instructions, list):
                 instructions.extend(_instructions)
-                
+
         # 3.1.1 Add instructions from the Model
         _model_instructions = self.model.get_instructions_for_model(self._tools_for_model)
         if _model_instructions is not None:

libs/agno/agno/knowledge/gcs/base.py

@@ -25,7 +25,7 @@ def gcs_blobs(self) -> List[storage.Blob]:
         if self.blob_name is not None:
             blobs_to_read.append(self.bucket.blob(self.blob_name))  # type: ignore
         elif self.prefix is not None:
-            blobs_to_read.extend(self.bucket.list_blobs(prefix=self.prefix))  # type: ignore 
+            blobs_to_read.extend(self.bucket.list_blobs(prefix=self.prefix))  # type: ignore
         else:
             blobs_to_read.extend(self.bucket.list_blobs())  # type: ignore
         return list(blobs_to_read)