| |
| """ |
| Reactive Echo Agent |
| |
| Implements a LangGraph-based ReAct loop where the LLM decides whether to |
| invoke tools and receives their results before continuing the conversation. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import json |
| import operator |
| from datetime import datetime |
| from pathlib import Path |
| from typing import Any, Dict, List, Optional, TypedDict, Annotated |
|
|
| from langchain_core.language_models import BaseLanguageModel |
| from langchain_core.messages import AnyMessage, SystemMessage, ToolMessage |
| from langchain_core.tools import BaseTool |
| from langgraph.graph import END, StateGraph |
|
|
|
|
| class ToolCallLog(TypedDict): |
| """Structured record of an executed tool call.""" |
|
|
| timestamp: str |
| tool_call_id: str |
| name: str |
| args: Any |
| content: str |
|
|
|
|
| class EchoAgentState(TypedDict): |
| """State carried through the LangGraph execution.""" |
|
|
| messages: Annotated[List[AnyMessage], operator.add] |
|
|
|
|
| class ReactiveEchoAgent: |
| """ |
| Minimal ReAct-style agent. |
| |
| The agent delegates decision making to the bound language model. Whenever |
| the model emits tool calls, the specified LangChain tools are executed and |
| their `ToolMessage` responses are appended to the conversation history |
| before handing control back to the model. |
| """ |
|
|
| def __init__( |
| self, |
| model: BaseLanguageModel, |
| tools: List[BaseTool], |
| *, |
| system_prompt: str = "", |
| checkpointer: Any = None, |
| log_tools: bool = True, |
| log_dir: Optional[str] = "logs", |
| ) -> None: |
| self._system_prompt = system_prompt |
| self._log_tools = log_tools |
| self._log_dir = Path(log_dir or "logs") |
| if self._log_tools: |
| self._log_dir.mkdir(parents=True, exist_ok=True) |
|
|
| |
| workflow = StateGraph(EchoAgentState) |
| workflow.add_node("process", self._process_request) |
| workflow.add_node("execute", self._execute_tools) |
| workflow.add_conditional_edges("process", self._has_tool_calls, {True: "execute", False: END}) |
| workflow.add_edge("execute", "process") |
| workflow.set_entry_point("process") |
|
|
| self.workflow = workflow.compile(checkpointer=checkpointer) |
| self.tools = {tool.name: tool for tool in tools} |
| self.model = model.bind_tools(list(self.tools.values())) |
|
|
| @property |
| def system_prompt(self) -> str: |
| return self._system_prompt |
|
|
| def update_system_prompt(self, prompt: str) -> None: |
| """Set a new system prompt for subsequent runs.""" |
| self._system_prompt = prompt |
|
|
| |
| def _process_request(self, state: Dict[str, Any]) -> Dict[str, List[AnyMessage]]: |
| messages: List[AnyMessage] = list(state.get("messages", [])) |
| if self._system_prompt: |
| messages = [SystemMessage(content=self._system_prompt)] + messages |
|
|
| response = self.model.invoke(messages) |
| return {"messages": [response]} |
|
|
| def _has_tool_calls(self, state: Dict[str, Any]) -> bool: |
| last_message = state["messages"][-1] |
| return bool(getattr(last_message, "tool_calls", [])) |
|
|
| def _execute_tools(self, state: Dict[str, Any]) -> Dict[str, List[ToolMessage]]: |
| tool_messages: List[ToolMessage] = [] |
| for call in state["messages"][-1].tool_calls: |
| tool_name = call.get("name") |
| tool_args = call.get("args", {}) |
| tool_id = call.get("id", "") |
|
|
| if tool_name not in self.tools: |
| result_content = json.dumps( |
| {"status": "error", "error": f"Unknown tool '{tool_name}'"}, ensure_ascii=False |
| ) |
| else: |
| try: |
| result = self.tools[tool_name].invoke(tool_args) |
| |
| result_content = json.dumps(result, ensure_ascii=False, default=str) |
| except Exception as exc: |
| result_content = json.dumps( |
| {"status": "error", "error": f"{type(exc).__name__}: {exc}"}, ensure_ascii=False |
| ) |
|
|
| message = ToolMessage( |
| tool_call_id=tool_id, |
| name=tool_name or "unknown_tool", |
| content=result_content, |
| additional_kwargs={"args": tool_args}, |
| ) |
| tool_messages.append(message) |
|
|
| self._log_tool_messages(tool_messages) |
| return {"messages": tool_messages} |
|
|
| |
| def _log_tool_messages(self, tool_messages: List[ToolMessage]) -> None: |
| if not self._log_tools or not tool_messages: |
| return |
|
|
| timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S") |
| log_path = self._log_dir / f"tool_calls_{timestamp}.json" |
| logs: List[ToolCallLog] = [] |
| for message in tool_messages: |
| logs.append(ToolCallLog( |
| tool_call_id=message.tool_call_id, |
| name=message.name, |
| args=message.additional_kwargs.get("args", {}), |
| content=message.content, |
| timestamp=datetime.utcnow().isoformat(), |
| )) |
|
|
| log_path.write_text(json.dumps(logs, indent=2), encoding="utf-8") |
|
|
