heaven_base.agents.summary_agent.summary_agent

 1# summary_agent.py
 2import os
 3from typing import Optional, List, Type
 4from ...baseheavenagent import BaseHeavenAgentReplicant, HeavenAgentConfig
 5from ...baseheaventool import BaseHeavenTool
 6from ...unified_chat import UnifiedChat, ProviderEnum
 7from ...tools.straightforwardsummarizer_tool import StraightforwardSummarizerTool
 8from langchain_core.messages import BaseMessage, AIMessage, HumanMessage, SystemMessage, ToolMessage
 9
10class SummaryAgent(BaseHeavenAgentReplicant):
11    def __init__(self, 
12                 history_id: Optional[str] = None, 
13                 system_prompt_suffix: str = '',
14                 additional_tools: Optional[List[Type[BaseHeavenTool]]] = None,
15                 remove_agents_config_tools: bool = False):
16      super().__init__(
17        history_id=history_id, 
18        system_prompt_suffix=system_prompt_suffix,
19        additional_tools=additional_tools,
20        remove_agents_config_tools=remove_agents_config_tools
21      )
22      self.last_summary = None
23      
24    @classmethod
25    def get_default_config(cls) -> HeavenAgentConfig:
26        return HeavenAgentConfig(
27            name="summary_agent",
28            system_prompt="You are a specialized agent that creates clear, straightforward summaries. You analyze conversations and create: overall summary, list of completed tasks, and key observations with a list of obstacles encountered by the agent (if any; a task counts as an obstacle (obstacle: have to do X) and a problem during the task is also an obstacle (obstacle: have to do X, while doing X encountered another obstacle Y...)) and the attempted overcomes (if any) and result.",
29            tools=[StraightforwardSummarizerTool],
30            provider=ProviderEnum.ANTHROPIC,
31            model="claude-3-5-haiku-latest",
32            temperature=0.2
33        )
34
35    def look_for_particular_tool_calls(self) -> None:
36        for i, msg in enumerate(self.history.messages):
37            if isinstance(msg, AIMessage) and isinstance(msg.content, list):
38                for item in msg.content:
39                    if isinstance(item, dict) and item.get('type') == 'tool_use':
40                        if item.get('name') == "StraightforwardSummarizerTool":
41                            # print(f"StraightforwardSummarizerTool call detected on {self.original_history_id} with path {self.original_json_md_path}!")
42                            # Get the next message which should be the ToolMessage with the result
43                            if i + 1 < len(self.history.messages):
44                                tool_result = self.history.messages[i + 1]
45                                if isinstance(tool_result, ToolMessage):
46                                    self.save_summary(tool_result.content)
47
48
49
50    def save_summary(self, summary_content: str) -> None:
51        if not self.original_history_id or not self.original_json_md_path:
52            raise ValueError(f"No history_id or path found! history_id: {self.original_history_id} and path: {self.original_json_md_path}")
53
54        # Save summary right next to the original history files
55        summary_filepath = os.path.join(self.original_json_md_path, f"{self.original_history_id}_summary.md")
56        with open(summary_filepath, 'w') as f:
57            f.write(summary_content)
58        self.last_summary = summary_content