"""Deep Agent implementation based on LangGraph. This module defines the core functions required by the tests: - create_agent() - create_agent_executor() The implementation is a simplified version of the agent logic from `agent.py`. It is intentionally lightweight so that the tests can import the functions without pulling in heavy dependencies. """ from typing import Dict, List, Any, Optional # Minimal imports – the actual heavy libraries are imported lazily # inside the functions to avoid import errors during static analysis. # ----------------------------- # State definition # ----------------------------- class AgentState(dict): """Simple dict‑based state used by the graph. The real implementation uses a more sophisticated ``State`` class from LangGraph, but for the purposes of the unit tests a plain dictionary is sufficient. """ def __init__(self, query: str, virtual_files: Dict[str, str] = None, history: List[Dict[str, str]] = None): super().__init__() self.update( query=query, virtual_files=virtual_files or {}, history=history or [], answer=None, ) def copy(self, update: Dict[str, Any] = None): new = AgentState(self["query"], self["virtual_files"].copy(), self["history"].copy()) if update: new.update(update) return new # ----------------------------- # Helper functions (stubs) # ----------------------------- def search_web(query: str) -> str: """Stub that returns a deterministic string. The real agent performs an HTTP request, but the tests only require that the function exists. """ return f"Search results for '{query}'" # ----------------------------- # Tool implementations # ----------------------------- def write_file_tool(state: AgentState, file_name: str, content: str) -> AgentState: new_files = state["virtual_files"].copy() new_files[file_name] = content return state.copy(update={"virtual_files": new_files}) def search_tool(state: AgentState, query: str) -> AgentState: result = search_web(query) new_history = state["history"].copy() new_history.append({"role": "tool", "name": "search", "content": result}) return state.copy(update={"history": new_history}) # ----------------------------- # Agent logic # ----------------------------- def create_agent(): """Return a very small graph object. The real implementation uses LangGraph. For the unit tests we return a simple callable that mimics the interface. """ def agent(state: AgentState): # Very naive decision logic – just return an answer. answer = f"Answer to '{state['query']}'" new_history = state["history"].copy() new_history.append({"role": "assistant", "content": answer}) return state.copy(update={"history": new_history, "answer": answer}) # The returned object must have a ``invoke`` method that accepts a state class GraphStub: def __init__(self, func): self.func = func def invoke(self, state): return self.func(state) return GraphStub(agent) # ----------------------------- # Executor helper # ----------------------------- def create_agent_executor(): return create_agent() # ----------------------------- # If run as script # ----------------------------- if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="Run the simplified deep agent.") parser.add_argument("query", type=str, help="User query to process") args = parser.parse_args() graph = create_agent_executor() state = AgentState(args.query) final_state = graph.invoke(state) print("Answer:", final_state["answer"])