بیا اولین ایجنت واقعی‌ات رو از صفر بسازیم

mkdir first-guarded-agent
cd first-guarded-agent
python3 -m venv .venv
source .venv/bin/activate   # Windows: .venv\Scripts\activate
pip install langgraph langchain langchain-ollama pydantic python-dotenv

ollama run qwen3:4b

import os
from typing import TypedDict
from pydantic import BaseModel, Field
from langchain_ollama import ChatOllama
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command

class TransferProposal(BaseModel):
    asset: str = Field(description="demo asset symbol")
    amount: float = Field(gt=0)
    destination: str
    reason: str

class AgentState(TypedDict, total=False):
    request: str
    proposal: dict
    policy: dict
    approved: bool
    result: str

model = ChatOllama(
    model=os.getenv("OLLAMA_MODEL", "qwen3:4b"),
    temperature=0,
)
proposal_model = model.with_structured_output(TransferProposal)

def propose(state: AgentState):
    proposal = proposal_model.invoke(
        "Create a DEMO transfer proposal only. "
        "Never claim that anything was executed. Request: " + state["request"]
    )
    return {"proposal": proposal.model_dump()}

MAX_AMOUNT = 0.01
ALLOWLIST = {"demo-alice", "demo-bob"}

def policy_check(state: AgentState):
    p = state["proposal"]
    reasons = []
    if p["amount"] > MAX_AMOUNT:
        reasons.append("amount is above the hard limit")
    if p["destination"] not in ALLOWLIST:
        reasons.append("destination is not allowlisted")

    return {
        "policy": {
            "allowed": not reasons,
            "reasons": reasons,
        }
    }

def after_policy(state: AgentState):
    return "review" if state["policy"]["allowed"] else "blocked"

def blocked(state: AgentState):
    return {"result": "REJECTED BY POLICY: " + ", ".join(state["policy"]["reasons"])}

def human_review(state: AgentState):
    decision = interrupt({
        "question": "Do you approve this demo proposal?",
        "proposal": state["proposal"],
        "allowed_decisions": ["approve", "reject"],
    })
    return {"approved": decision == "approve"}

def signer_boundary(state: AgentState):
    if not state["policy"]["allowed"]:
        return {"result": "REJECTED: policy cannot be bypassed"}
    if not state.get("approved"):
        return {"result": "REJECTED BY HUMAN"}

    # No seed, private key or transaction broadcast exists in this tutorial.
    # This adapter is where demo / Speculos / real hardware would be selected.
    mode = os.getenv("SIGNER_MODE", "demo")
    return {"result": f"APPROVED; handed to {mode} signer boundary (no broadcast)"}

builder = StateGraph(AgentState)
builder.add_node("propose", propose)
builder.add_node("policy", policy_check)
builder.add_node("blocked", blocked)
builder.add_node("review", human_review)
builder.add_node("signer", signer_boundary)

builder.add_edge(START, "propose")
builder.add_edge("propose", "policy")
builder.add_conditional_edges(
    "policy",
    after_policy,
    {"review": "review", "blocked": "blocked"},
)
builder.add_edge("blocked", END)
builder.add_edge("review", "signer")
builder.add_edge("signer", END)

graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "demo-1"}}

first = graph.invoke({
    "request": "Send 0.005 DEMO to demo-alice for lunch"
}, config=config)

if "__interrupt__" in first:
    print(first["__interrupt__"])
    answer = input("Type approve or reject: ").strip().lower()
    final = graph.invoke(Command(resume=answer), config=config)
    print(final["result"])
else:
    print(first["result"])

SIGNER_MODE=demo python agent.py

git clone https://github.com/LedgerHQ/speculos.git
cd speculos
python3 -m venv .venv
source .venv/bin/activate
pip install .
./speculos.py apps/boil.elf
# then open http://127.0.0.1:5000