Skip to content

feat(dynaflow): conversation / short-term memory store on DynamoDB (#290) #336

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
shivamm-gupta wants to merge 1 commit into
base: development
Choose a base branch
from
Open

feat(dynaflow): conversation / short-term memory store on DynamoDB (#290) #336

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 17 additions & 0 deletions src/dynavec/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,15 @@
run_context,
set_default_run_logger,
)
from .memory import (
BaseMemory,
DynamoDBMemoryStore,
InMemoryMemoryStore,
MemoryPolicy,
SummaryMemoryPolicy,
TokenBudgetMemoryPolicy,
WindowMemoryPolicy,
)
from .models import (
Document,
ExplainedSearchResult,
Expand Down Expand Up @@ -189,6 +198,14 @@
"BaseCheckpointer",
"MemoryCheckpointer",
"DynamoDBCheckpointer",
# Conversation Memory (dynaflow)
"BaseMemory",
"InMemoryMemoryStore",
"DynamoDBMemoryStore",
"MemoryPolicy",
"WindowMemoryPolicy",
"TokenBudgetMemoryPolicy",
"SummaryMemoryPolicy",
# Ingestion
"Record",
"ingest",
Expand Down
16 changes: 16 additions & 0 deletions src/dynavec/agents/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,15 @@
DynamoDBCheckpointer,
MemoryCheckpointer,
)
from ..memory import (
BaseMemory,
DynamoDBMemoryStore,
InMemoryMemoryStore,
MemoryPolicy,
SummaryMemoryPolicy,
TokenBudgetMemoryPolicy,
WindowMemoryPolicy,
)
from .base import (
AgentResult,
AgentStep,
Expand Down Expand Up @@ -45,4 +54,11 @@
"BaseCheckpointer",
"MemoryCheckpointer",
"DynamoDBCheckpointer",
"BaseMemory",
"InMemoryMemoryStore",
"DynamoDBMemoryStore",
"MemoryPolicy",
"WindowMemoryPolicy",
"TokenBudgetMemoryPolicy",
"SummaryMemoryPolicy",
]
83 changes: 63 additions & 20 deletions src/dynavec/agents/react.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@

if TYPE_CHECKING:
from ..checkpoint import BaseCheckpointer
from ..memory import BaseMemory

DEFAULT_REACT_SYSTEM_PROMPT = """You are a helpful and precise reasoning agent.
You solve tasks step-by-step using a ReAct (Reason + Act) approach.
Expand Down Expand Up @@ -76,13 +77,15 @@ def __init__(
system_prompt: str | None = None,
max_steps: int = 10,
checkpointer: BaseCheckpointer | None = None,
memory: BaseMemory | None = None,
) -> None:
self.model = model
self.system_prompt = (
system_prompt if system_prompt is not None else DEFAULT_REACT_SYSTEM_PROMPT
)
self.max_steps = max(1, max_steps)
self.checkpointer = checkpointer
self.memory = memory

self._tools_map: dict[str, AgentTool] = {}
self._chat_tools: list[Tool] = []
Expand All @@ -93,9 +96,7 @@ def __init__(
self._chat_tools.append(t.to_chat_tool())
elif tools:
for raw_tool in tools:
agent_tool = (
raw_tool if isinstance(raw_tool, AgentTool) else AgentTool(raw_tool)
)
agent_tool = raw_tool if isinstance(raw_tool, AgentTool) else AgentTool(raw_tool)
self._tools_map[agent_tool.name] = agent_tool
self._chat_tools.append(agent_tool.to_chat_tool())

Expand All @@ -118,9 +119,7 @@ async def _adispatch_tool_call(self, tc: ToolCall) -> str:
return await tool_instance.aexecute(tc.arguments)
return f"Error: Tool {tc.name!r} is not registered in available tools."

def _process_tool_calls(
self, tool_calls: list[ToolCall], messages: list[Message]
) -> list[str]:
def _process_tool_calls(self, tool_calls: list[ToolCall], messages: list[Message]) -> list[str]:
step_observations: list[str] = []
for tc in tool_calls:
obs = self._dispatch_tool_call(tc)
Expand Down Expand Up @@ -150,12 +149,9 @@ async def _aprocess_tool_calls(
)
return step_observations

def _finalize_result(
self, steps: list[AgentStep], total_tool_calls: int
) -> AgentResult:
def _finalize_result(self, steps: list[AgentStep], total_tool_calls: int) -> AgentResult:
last_output = (
steps[-1].thought
or (steps[-1].observations[-1] if steps[-1].observations else "")
steps[-1].thought or (steps[-1].observations[-1] if steps[-1].observations else "")
if steps
else ""
)
Expand Down Expand Up @@ -194,7 +190,21 @@ def run(
start_step = existing_cp.step + 1

if not messages:
messages = self._init_messages(goal)
if self.memory is not None and thread_id is not None:
existing = self.memory.get_messages(thread_id)
if not existing and self.system_prompt:
self.memory.append(
thread_id,
Message(role="system", content=self.system_prompt),
)
self.memory.append(
thread_id,
Message(role="user", content=goal),
)
messages = list(self.memory.get_messages(thread_id))
else:
messages = self._init_messages(goal)

if self.checkpointer is not None and thread_id is not None:
self.checkpointer.put(
thread_id,
Expand Down Expand Up @@ -225,6 +235,12 @@ def run(
observations=[],
)
steps.append(step)
out = msg.content or ""
if self.memory is not None and thread_id is not None:
self.memory.append(
thread_id,
Message(role="assistant", content=out),
)
if self.checkpointer is not None and thread_id is not None:
self.checkpointer.put(
thread_id,
Expand All @@ -233,14 +249,14 @@ def run(
"messages": [_message_to_dict(m) for m in messages],
"steps": [_step_to_dict(s) for s in steps],
"total_tool_calls": total_tool_calls,
"output": msg.content or "",
"output": out,
"finished": True,
},
node_id=f"step_{step_idx}",
step=step_idx,
)
return AgentResult(
output=msg.content or "",
output=out,
steps=steps,
finished=True,
termination_reason="completed",
Expand Down Expand Up @@ -274,6 +290,11 @@ def run(
)

res = self._finalize_result(steps, total_tool_calls)
if self.memory is not None and thread_id is not None:
self.memory.append(
thread_id,
Message(role="assistant", content=res.output),
)
if self.checkpointer is not None and thread_id is not None:
self.checkpointer.put(
thread_id,
Expand Down Expand Up @@ -316,7 +337,20 @@ async def arun(
start_step = existing_cp.step + 1

if not messages:
messages = self._init_messages(goal)
if self.memory is not None and thread_id is not None:
existing = self.memory.get_messages(thread_id)
if not existing and self.system_prompt:
self.memory.append(
thread_id,
Message(role="system", content=self.system_prompt),
)
self.memory.append(
thread_id,
Message(role="user", content=goal),
)
messages = list(self.memory.get_messages(thread_id))
else:
messages = self._init_messages(goal)
if self.checkpointer is not None and thread_id is not None:
self.checkpointer.put(
thread_id,
Expand Down Expand Up @@ -347,6 +381,12 @@ async def arun(
observations=[],
)
steps.append(step)
out = msg.content or ""
if self.memory is not None and thread_id is not None:
self.memory.append(
thread_id,
Message(role="assistant", content=out),
)
if self.checkpointer is not None and thread_id is not None:
self.checkpointer.put(
thread_id,
Expand All @@ -355,14 +395,14 @@ async def arun(
"messages": [_message_to_dict(m) for m in messages],
"steps": [_step_to_dict(s) for s in steps],
"total_tool_calls": total_tool_calls,
"output": msg.content or "",
"output": out,
"finished": True,
},
node_id=f"step_{step_idx}",
step=step_idx,
)
return AgentResult(
output=msg.content or "",
output=out,
steps=steps,
finished=True,
termination_reason="completed",
Expand All @@ -371,9 +411,7 @@ async def arun(
)

total_tool_calls += len(msg.tool_calls)
step_observations = await self._aprocess_tool_calls(
list(msg.tool_calls), messages
)
step_observations = await self._aprocess_tool_calls(list(msg.tool_calls), messages)

step = AgentStep(
step_number=step_idx,
Expand All @@ -398,6 +436,11 @@ async def arun(
)

res = self._finalize_result(steps, total_tool_calls)
if self.memory is not None and thread_id is not None:
self.memory.append(
thread_id,
Message(role="assistant", content=res.output),
)
if self.checkpointer is not None and thread_id is not None:
self.checkpointer.put(
thread_id,
Expand Down
Loading
Loading