-
-
Notifications
You must be signed in to change notification settings - Fork 68
Agent System
Judah Paul edited this page Feb 16, 2026
·
4 revisions
The Agent System is the brain of GPT Home, powered by LangGraph for orchestration and LiteLLM for model abstraction. This page covers the agent architecture, configuration, state management, and extension points.
┌──────────────────────────────────────────────────────────────────┐
│ GPTHomeAgent │
├──────────────────────────────────────────────────────────────────┤
│ │
│ ┌────────────────┐ ┌────────────────────────────────┐ │
│ │ AgentConfig │───────▶│ LangGraph ReAct Agent │ │
│ │ │ │ │ │
│ │ • model │ │ ┌──────────────────────────┐ │ │
│ │ • temperature │ │ │ System Prompt │ │ │
│ │ • max_tokens │ │ │ + Memory Context │ │ │
│ │ • custom_inst │ │ └──────────────────────────┘ │ │
│ │ • db_url │ │ │ │ │
│ └────────────────┘ │ ▼ │ │
│ │ ┌──────────────────────────┐ │ │
│ ┌────────────────┐ │ │ ChatLiteLLM │ │ │
│ │ Checkpointer │───────▶│ │ (100+ providers) │ │ │
│ │ (PostgreSQL) │ │ └──────────────────────────┘ │ │
│ └────────────────┘ │ │ │ │
│ │ ▼ │ │
│ ┌────────────────┐ │ ┌──────────────────────────┐ │ │
│ │ Memory Store │───────▶│ │ Tools + Memory │ │ │
│ │ (pgvector) │ │ │ Tools │ │ │
│ └────────────────┘ │ └──────────────────────────┘ │ │
│ └────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────────┘
The main agent implementation in src/agent/core.py:
class GPTHomeAgent(BaseAgent):
"""Main agent implementation using Strategy pattern for tool selection."""
def __init__(
self,
config: AgentConfig,
checkpointer: Optional[AsyncPostgresSaver] = None,
store: Optional[BaseStore] = None
):
self.config = config
self.checkpointer = checkpointer # Conversation persistence
self.store = store # Memory storage
self.tool_registry = ToolRegistry()
self._agent = None
self._initialized = False| Method | Description |
|---|---|
initialize() |
Lazy initialization of LangGraph agent |
invoke(text, user_id, thread_id) |
Process user input synchronously |
stream(text, user_id, thread_id) |
Stream responses for real-time output |
_build_system_prompt(state) |
Construct prompt with memory context |
Configuration management with Builder pattern in src/agent/config.py:
@dataclass
class AgentConfig:
"""Configuration for the GPT Home agent."""
model: str = "gpt-4o-mini" # LiteLLM model name
temperature: float = 0.7 # Response creativity
max_tokens: int = 1024 # Max response length
custom_instructions: str = "" # User-defined behavior
embedding_model: str = "openai:text-embedding-3-small" # Format: provider:model
embedding_dims: int = 1536 # Vector dimensions
database_url: Optional[str] = None # PostgreSQL connectionConfiguration is loaded from multiple sources (priority order):
- Environment variables (
MODEL,DATABASE_URL) -
settings.jsonfile - Default values
@classmethod
def from_settings(cls, settings_path: Optional[Path] = None) -> "AgentConfig":
"""Factory method to create config from settings file."""
config_data = {
"model": os.getenv("MODEL") or settings.get("model", "gpt-4o-mini"),
"temperature": settings.get("temperature", 0.7),
"max_tokens": settings.get("max_tokens", 1024),
"custom_instructions": settings.get("custom_instructions", ""),
}
config_data["database_url"] = os.getenv("DATABASE_URL")
return cls(**config_data)# Fluent configuration
config = AgentConfig.builder() \
.with_model("claude-3-haiku-20240307") \
.with_temperature(0.5) \
.with_max_tokens(2048) \
.with_custom_instructions("Always respond in a friendly tone.") \
.with_database_url("postgresql://...") \
.with_embedding("text-embedding-3-large", 3072) \
.build()State schema for the agent in src/agent/state.py:
class AgentState(MessagesState):
"""State schema for the GPT Home agent.
Extends MessagesState with additional fields for memory and context.
Uses reducer pattern for message accumulation.
"""
user_id: str # User identifier for memory namespacing
thread_id: str # Conversation thread ID
memories: Annotated[list[dict], add] # Accumulated memories
context: dict # Additional contextGPT Home uses LangGraph's create_react_agent for a ReAct (Reasoning + Acting) loop:
┌───────────────────────────────────────────────────────────┐
│ ReAct Loop │
├───────────────────────────────────────────────────────────┤
│ │
│ ┌─────────┐ ┌─────────┐ ┌─────────────┐ │
│ │ START │────▶│ THINK │────▶ │ DECIDE │ │
│ └─────────┘ │ │ │ │ │
│ │ Analyze │ │ Tool needed?│ │
│ │ request │ │ │ │
│ └─────────┘ └──────┬──────┘ │
│ │ │
│ ┌──────────────────┼────────┐ │
│ │ Yes │ No │ │
│ ▼ ▼ │ │
│ ┌─────────────┐ ┌───────────┐ │ │
│ │ ACTION │ │ RESPOND │ │ │
│ │ │ │ │ │ │
│ │ Execute │ │ Generate │──┼──▶│ END
│ │ tool │ │ response │ │ │
│ └──────┬──────┘ └───────────┘ │ │
│ │ │ │
│ ▼ │ │
│ ┌─────────────┐ │ │
│ │ OBSERVE │ │ │
│ │ │────────────────────┘ │
│ │ Process │ │
│ │ tool output │ │
│ └─────────────┘ │
│ │
└───────────────────────────────────────────────────────────┘
async def initialize(self):
"""Lazy initialization of the agent graph."""
if self._initialized:
return
# LiteLLM wraps 100+ providers with unified interface
llm = ChatLiteLLM(
model=self.config.model,
temperature=self.config.temperature,
max_tokens=self.config.max_tokens,
api_key=os.getenv("LITELLM_API_KEY"),
)
# Combine action tools and memory tools
action_tools = get_all_tools() # weather, spotify, lights, etc.
memory_tools = [
create_manage_memory_tool(namespace=("memories", "{user_id}")),
create_search_memory_tool(namespace=("memories", "{user_id}")),
]
self._agent = create_react_agent(
llm,
tools=action_tools + memory_tools,
prompt=self._build_system_prompt,
checkpointer=self.checkpointer, # PostgreSQL persistence
store=self.store, # Memory storage
)The system prompt is dynamically built with memory context:
def _build_system_prompt(self, state: AgentState) -> list:
"""Build system prompt with memory context."""
# Search for relevant memories
memories_text = ""
if self.store and state.get("user_id"):
memories = store.search(
("memories", state["user_id"]),
query=state["messages"][-1].content,
limit=5
)
if memories:
memories_text = "\n".join([
f"- {m.value.get('content', m.value)}"
for m in memories
])
# Build integration status from service_status kwarg
statuses = getattr(self, "_current_service_status", {})
available = [name for name, ok in statuses.items() if ok]
if available:
integrations_text = "Available integrations: " + ", ".join(available) + "."
else:
integrations_text = "No integrations are configured yet."
system_content = f"""You are a helpful AI assistant for GPT Home...
## Integration Status
{integrations_text}
## Important Tool Usage Guidelines
- For simple greetings or general conversation, respond naturally
without mentioning integrations.
- When users ask about weather, call the weather tool immediately.
- When users ask to play music, call the Spotify tool directly.
- Be proactive - call tools first, ask clarifying questions
only if the tool fails.
## User Memories
<memories>
{memories_text if memories_text else "No memories stored yet."}
</memories>
When users share preferences, use the memory tools to save them.
Be concise as responses will be spoken aloud."""
return [{"role": "system", "content": system_content}, *state["messages"]]async def invoke(self, text: str, user_id: str = "default",
thread_id: str = "default", **kwargs) -> str:
"""Process user input and return response."""
await self.initialize()
# Service statuses are stored for system prompt injection
# (not appended to user message, to avoid polluting simple queries)
self._current_service_status = kwargs.get("service_status", {})
config: RunnableConfig = {
"configurable": {
"thread_id": thread_id,
"user_id": user_id,
}
}
result = await self._agent.ainvoke(
{"messages": [{"role": "user", "content": text}]},
config=config
)
if result and result.get("messages"):
return result["messages"][-1].content
return "I'm sorry, I couldn't process that request."async def stream(self, text: str, user_id: str = "default",
thread_id: str = "default", **kwargs):
"""Stream responses for real-time output."""
await self.initialize()
async for event in self._agent.astream_events(
{"messages": [{"role": "user", "content": text}]},
config={"configurable": {"thread_id": thread_id, "user_id": user_id}},
version="v2"
):
yield eventGPT Home uses LiteLLM to support 100+ AI providers with a unified interface:
┌────────────────────────────────────────────────────────────┐
│ LiteLLM Layer │
├────────────────────────────────────────────────────────────┤
│ │
│ LITELLM_API_KEY environment variable │
│ │
│ Model name determines provider: │
│ │
│ ┌─────────────────┬─────────────────────────────────┐ │
│ │ Model Prefix │ Provider │ │
│ ├─────────────────┼─────────────────────────────────┤ │
│ │ gpt-* │ OpenAI │ │
│ │ claude-* │ Anthropic │ │
│ │ gemini/* │ Google │ │
│ │ command-* │ Cohere │ │
│ │ mistral/* │ Mistral AI │ │
│ │ ollama/* │ Ollama (local) │ │
│ │ groq/* │ Groq │ │
│ │ together_ai/* │ Together AI │ │
│ └─────────────────┴─────────────────────────────────┘ │
│ │
└────────────────────────────────────────────────────────────┘
# OpenAI (default)
MODEL=gpt-4o-mini
LITELLM_API_KEY=sk-...
# Anthropic Claude
MODEL=claude-3-haiku-20240307
LITELLM_API_KEY=sk-ant-...
# Google Gemini
MODEL=gemini/gemini-1.5-flash
LITELLM_API_KEY=...
# Local Ollama
MODEL=ollama/llama3.2
# No API key needed for localLangGraph checkpointing enables conversation continuity:
# Initialize PostgreSQL checkpointer
checkpointer = AsyncPostgresSaver(conn=connection_pool)
await checkpointer.setup()
# Each thread_id maintains its own conversation history
# User can continue conversations across sessionsUser: "What's the weather?" thread_id: "session_user1"
│
▼
┌─────────────────┐
│ Agent Process │──────▶ Response: "It's 72°F and sunny"
└────────┬────────┘
│
▼
┌─────────────────┐
│ Save Checkpoint │──────▶ PostgreSQL (checkpoints table)
│ │
│ • messages[] │
│ • tool calls │
│ • tool results │
└─────────────────┘
User: "And tomorrow?" thread_id: "session_user1"
│
▼
┌─────────────────┐
│ Load Checkpoint │◀────── PostgreSQL
│ │
│ Context: Prior │
│ weather query │
└────────┬────────┘
│
▼
Agent understands "tomorrow" refers to weather forecast
The create_agent factory handles dependency injection:
async def create_agent(config: Optional[AgentConfig] = None) -> GPTHomeAgent:
"""Factory function to create and initialize an agent with all dependencies."""
if config is None:
config = AgentConfig.from_settings()
checkpointer = None
store = None
if config.database_url:
try:
# Initialize PostgreSQL checkpointer
async with AsyncPostgresSaver.from_conn_string(config.database_url) as saver:
await saver.setup()
checkpointer = saver
except Exception as e:
print(f"Warning: Could not initialize checkpointer: {e}")
try:
# Initialize memory store with vector index
store = AsyncPostgresStore.from_conn_string(
config.database_url,
index={
"dims": config.embedding_dims,
"embed": config.embedding_model,
"fields": ["content", "$"],
}
)
await store.setup()
except Exception as e:
print(f"Warning: Could not initialize store: {e}")
agent = GPTHomeAgent(config, checkpointer, store)
await agent.initialize()
return agentVia settings.json:
{
"model": "gpt-4o-mini",
"temperature": 0.7,
"custom_instructions": "Always greet the user warmly. Prefer metric units. Speak in a friendly, casual tone."
}- Create tool in
src/tools/your_tool.py:
from langchain_core.tools import tool
@tool
async def your_tool(query: str) -> str:
"""Your tool description for the LLM.
Args:
query: What the user is asking for
Returns:
Result string
"""
# Implementation
return "Result"- Register in
src/tools/__init__.py:
from .your_tool import your_tool
__all__ = [..., "your_tool"]- Add to registry in
src/tools/registry.py:
from .your_tool import your_tool
registry.register(
your_tool,
ToolMetadata(
name="your_tool",
description="Tool description",
category="productivity",
requires_api_key=True,
api_key_env_var="YOUR_API_KEY"
)
)- Learn about the Memory System for persistent context
- Explore all Tools available to the agent
- See Configuration for all options