Agent Architecture

from langchain_core.tools import tool
from pydantic import BaseModel, Field
 
# Simple tool — args inferred from function signature
@tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression and return the result."""
    return str(eval(expression))  # Use a sandbox in production
 
# Structured tool — explicit schema with validation
class EmailInput(BaseModel):
    to: str = Field(description="Recipient email address")
    subject: str = Field(description="Email subject line")
    body: str = Field(description="Email body content")
 
@tool(args_schema=EmailInput)
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email to the specified recipient."""
    mail_service.send(to=to, subject=subject, body=body)
    return f"Email sent to {to}"

from langchain_core.tools import tool, ToolException
 
@tool(handle_tool_error=True)
def query_database(sql: str) -> str:
    """Run a read-only SQL query against the analytics database."""
    if "DROP" in sql.upper() or "DELETE" in sql.upper():
        raise ToolException("Destructive queries are not allowed.")
    try:
        rows = db.execute(text(sql)).fetchall()
        return json.dumps([dict(r) for r in rows])
    except Exception as e:
        raise ToolException(f"Query failed: {str(e)}")

const agent = agent<AgentState>({
  assistantId: 'react_agent',
});
 
// The error() signal captures both transport and agent errors
const error = computed(() => agent.error());
 
// In your template
@Component({
  changeDetection: ChangeDetectionStrategy.OnPush,
  template: `
    @if (error()) {
      <app-error-banner [error]="error()" (retry)="retry()" />
    }
  `,
})
export class AgentComponent {
  error = computed(() => this.agent.error());
 
  retry() {
    // Re-submit the last message to retry
    this.agent.submit(this.lastInput);
  }
}

from langgraph.checkpoint.postgres import PostgresSaver
 
checkpointer = PostgresSaver.from_connection_string(DATABASE_URL)
graph = builder.compile(checkpointer=checkpointer)
 
# Every node execution creates a checkpoint:
# checkpoint_1: after "model" (LLM decided to call search_docs)
# checkpoint_2: after "tools" (search_docs returned results)
# checkpoint_3: after "model" (LLM responded with final answer)

const agent = agent<AgentState>({
  assistantId: 'react_agent',
  threadId: signal('thread_abc123'),
});
 
// Full checkpoint timeline — every state snapshot
const timeline = computed(() => agent.history());
 
// Current branch (for time-travel)
const branch = computed(() => agent.branch());

@Component({
  selector: 'app-debug-timeline',
  changeDetection: ChangeDetectionStrategy.OnPush,
  template: `
    <div class="timeline">
      @for (checkpoint of history(); track checkpoint.id) {
        <button
          [class.active]="checkpoint.id === currentCheckpoint()"
          (click)="timeTravel(checkpoint.id)">
          <span class="node">{{ checkpoint.node }}</span>
          <span class="time">{{ checkpoint.timestamp | date:'medium' }}</span>
        </button>
      }
    </div>
 
    <div class="state-inspector">
      <h3>State at checkpoint</h3>
      <pre>{{ selectedState() | json }}</pre>
    </div>
  `,
})
export class DebugTimelineComponent {
  history = computed(() => this.agent.history());
  currentCheckpoint = signal<string | null>(null);
 
  selectedState = computed(() => {
    const id = this.currentCheckpoint();
    return this.history().find(c => c.id === id)?.values;
  });
 
  timeTravel(checkpointId: string) {
    this.currentCheckpoint.set(checkpointId);
    this.agent.submit({}, { checkpointId });
  }
}

# Simple, powerful, covers 80% of use cases
builder = StateGraph(AgentState)
builder.add_node("model", call_model)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "model")
builder.add_conditional_edges("model", should_continue)
builder.add_edge("tools", "model")
graph = builder.compile()

from langgraph.types import Interrupt
 
def propose_action(state: AgentState) -> dict:
    plan = llm.invoke(state["messages"])
    raise Interrupt(value={"action": plan.content, "requires_approval": True})
 
def execute_action(state: AgentState) -> dict:
    # Only runs after human approves
    return perform_action(state["pending_action"])

builder = StateGraph(OrchestratorState)
builder.add_node("supervisor", supervisor)
builder.add_node("researcher", researcher_subgraph)
builder.add_node("analyst", analyst_subgraph)
builder.add_conditional_edges("supervisor", route_to_agent)