03 · Agent Pipeline

Third installment of the architecture deep-dive: Agent lifecycle, ConversationSession tool loop, DialogueManager RAG composition, Transferer cross-agent dispatch, tool system, and client abstraction. Corresponds to ARCHITECTURE.md §4.

---

§4.1 Why Not Use an Off-the-Shelf Agent Framework

Framework	Reason for rejection
LangChain	Abstraction layer too thick; cannot finely control single-agent lifecycle, tool approval timing, or memory write points
AutoGen	Multi-agent collaboration model does not match Familiars' three-tier architecture (Manager → Assistant → Dialogue)
CrewAI	Local small models (decider / rewriter / memorizer-writer) need deep integration with the main loop

The custom pipeline treats "plan → tool call → memory → report" as first-class citizens, with explicit interception points at every stage.

§4.2 Agent Class (src/models/assistant.py)

4.2.1 Assistant Construction

class Assistant:
    def __init__(self, id: int, mode='normal', description: str = '',
                 name: str = '', notification_type: str = 'SYSTEM',
                 path_override: Optional[str] = None):
        self.id = id
        self.workspace_route = _resolve_workspace_root()
        
        # Path resolution
        if path_override:
            # Program worker: absolute path, e.g. <repo>/runs/programs/<pid>/agents/<role>
            self.path = path_override
            _agent_dir = os.path.abspath(path_override)
            _agents_parent = os.path.dirname(_agent_dir)
            _memorizer_root = os.path.dirname(_agents_parent)
            _agent_key = os.path.basename(_agent_dir)
            self.memorizer = Memorizer(_memorizer_root, agent_id=_agent_key)
        else:
            self.path = f'{self.workspace_route}/workspace/agents/agent_{self.id}'
            self.memorizer = Memorizer(
                os.path.join(self.workspace_route, "workspace"),
                agent_id=f"agent_{self.id}"
            )
        
        # Core components
        self.vision_processor = VisionProcessor()  # MultimodalTools
        self.dialogue_manager = DialogueManager(
            self.memorizer,
            agent_path=self.path,
            top_k=int(os.getenv("TOP_K") or 5),
            workspace_route=self.workspace_route
        )
        
        # Auto-detect notification type
        if self.notification_type == 'SYSTEM':
            if self.info['os'] == 'Darwin':
                self.notification_type = 'MacOS'
            elif self.info['os'] == 'Windows':
                self.notification_type = 'Windows'
            else:
                self.notification_type = 'DESKTOP'

4.2.2 AssistantManager

class AssistantManager:
    """Orchestrator for N Assistants.
    
    Responsibilities:
    - Agent lifecycle (create / delete / list)
    - APScheduler cron-driven report aggregation
    - Separate MANAGER_MODEL for executive summaries + persona evolution
    """

4.2.3 Signal Evaluation (Decider Integration)

def evaluate_signal(
    self,
    *,
    trigger_type: str,           # "new_message" / "external_info" / ...
    trigger_summary: str,         # Trigger summary
    trigger_details: Optional[dict] = None,
    memory: Optional[list[str]] = None,
    user_state: Optional[dict] = None,
    recent_user_messages: Optional[list[str]] = None,
):
    """Ask the decider whether this agent should act on a signal.
    
    Returns a DecisionOutcome (fire: bool, action_category: str, act_score: float).
    Always returns — never raises. Worst case: fire=False with source indicating why.
    """
    from src.decider import DeciderService
    svc = DeciderService.shared()
    return svc.evaluate(
        agent_id=self.id,
        trigger_type=trigger_type,
        trigger_summary=trigger_summary,
        ...
    )

4.2.4 Proactive Messages

def send_proactive_message(
    self,
    *,
    content: str,
    signal_summary: str,
    action_category: str,
    act_score: Optional[float] = None,
    decision_id: Optional[str] = None,
    dialogue_id: int = 0,
    title: Optional[str] = None,
    fired_at: Optional[float] = None,
    extra: Optional[dict] = None,
) -> Optional[int]:
    """Attach an agent-initiated A-node to dialogue_id and persist.
    
    origin='proactive' + metadata → frontend rating UI can POST feedback.
    Returns the new node id, or None if no dialogue exists.
    """

4.2.5 Notification Channels

Channel	Implementation	Description
Windows	win10toast.ToastNotifier	Native toast; falls back to desktop_notify on failure
macOS	osascript -e 'display notification ...'	Notification Center
Desktop	src.services.desktop_notify.notify_desktop	Electron renderer → OS toast
Telegram	requests.post(api.telegram.org/bot{token}/sendMessage)	Remote push

§4.3 ConversationSession (src/models/conversation.py)

4.3.1 Constructor Parameters

class ConversationSession:
    def __init__(
        self,
        model: Optional[str] = None,        # Default: select_model(fallback_env="DEFAULT_MODEL")
        timeout: Optional[int] = None,       # Default: TIMEOUT=300s
        tools: Optional[list[dict]] = None,  # OpenAI function schemas
        system_prompt: str = "",
        base_url: Optional[str] = None,
        api_key: Optional[str] = None,
        dispatcher: Optional[ToolDispatcher] = None,
    ):

4.3.2 CircuitBreaker

class CircuitBreaker:
    """Per-URL circuit breaker, three states: CLOSED → OPEN → HALF_OPEN.
    
    CLOSED: requests flow normally, counting failures.
    OPEN: all requests immediately rejected for recovery_seconds.
    HALF_OPEN: allows one probe request; success → CLOSED, failure → OPEN.
    """
    
    def __init__(
        self,
        failure_threshold: int = 5,         # CB_FAILURE_THRESHOLD
        recovery_seconds: float = 30.0,     # CB_RECOVERY_SECONDS
        half_open_max: int = 1,
    ):

Globally shared: _breakers: dict[str, CircuitBreaker] indexed by base URL; all ConversationSession instances share the same breaker.

4.3.3 Retry Logic

_RETRYABLE_STATUS = {408, 429, 500, 502, 503, 504}

# Permanent content-policy rejections (non-retryable)
_PERMANENT_CONTENT_ERROR_CODES = frozenset({
    "local:sensitive_words",           # api3.wlai.vip Chinese content filter
    "content_filter",                  # Azure OpenAI
    "content_policy_violation",        # OpenAI
    "safety",                          # Gemini / Vertex
    "prohibited_content",              # Gemini
})

4.3.4 Streaming Decision

_REASONING_MODEL_PREFIXES = ("gpt-5", "gpt-o", "o1-", "o3-", "o4-")
_REASONING_MODEL_EXACT = frozenset({"o1", "o3", "o4"})

def _should_stream(model: str) -> bool:
    """Returns False for reasoning models (gpt-5.x, gpt-o1, gpt-o3).
    
    Reason: reasoning-model SSE delta protocol is incompatible with the
    existing assembler — HTTP 200 but prompt_tokens=0, completion_tokens=0,
    100–200s latency.
    
    Environment variable overrides:
    FAMILIARS_STREAM_LLM=true  → force enable (debugging)
    FAMILIARS_STREAM_LLM=false → force disable
    """

4.3.5 Tool Loop (contact() method)

async def contact(self, history: list[dict], **kwargs) -> dict:
    """Run the tool-call loop on top of chat().
    
    Each turn the LLM may issue ≥1 tool calls → dispatch → tool_result → continue loop.
    Until max_turns or the LLM stops issuing tool calls.
    
    Turn budget escalation: when budget is exhausted, asks the manager LLM whether to extend.
    """

4.3.6 Auto Document Refresh

def _build_auto_refresh_messages(path_norm: str, write_tool: str,
                                  progress_footer: str = "") -> list[dict]:
    """Synthesize (assistant tool_call, tool result) pair after a write operation.
    
    On the next mutation of the same file, _mark_stale_docx_reads collapses old content.
    Returns two messages ready to append to history.
    """

§4.4 DialogueManager

class DialogueManager:
    """Manages N Dialogues (dialogue trees + history).
    
    Responsibilities:
    - _load_existing_dialogues() reads from disk
    - store_dialogue(dialogue_id) persists
    - build_rag_context(query) assembles RAG context
    - release_memory() releases resident memory (JIT load on demand)
    """
    
    def __init__(self, memorizer, agent_path, top_k=5, workspace_route=""):
        self.memorizer = memorizer
        self.agent_path = agent_path
        self.top_k = top_k
        self.dialogue: list[Dialogue] = []

§4.5 Transferer (Cross-Agent Dispatch)

src/models/transferer.py is the single chokepoint for all cross-agent calls.

def dispose(request_model: str, request: str, **kwargs) -> Any:
    """Resolve the destination string to a callable.
    
    Priority:
    1. Tool client aliases — "BrowseTool_Client", "browser_tools" → instantiate ToolClient
    2. Tool method paths — "BaseTools.open_app" → resolve to src.utils.* methods
    3. Model names — send directly to chat completion (no tool schemas)
    4. Empty / "Writer" / unknown → fall back to DEFAULT_MODEL
    """

Client Registry

_CLIENT_REGISTRY: dict[str, str] = {
    "basetoolclient":     "src.models.clients.base_client:BaseToolClient",
    "browsertoolclient":  "src.models.clients.browser_client:BrowserToolClient",
    "webdive":            "src.models.clients.browser_client:WebDiveToolClient",
    "multimodaltoolclient": "src.models.clients.multimodal_client:MultimodalToolClient",
    "inputtoolclient":    "src.models.clients.input_client:InputToolClient",
    "doctoolclient":      "src.models.clients.doc_client:DocToolClient",
    "dailytoolclient":    "src.models.clients.daily_client:DailyToolClient",
    "systemtoolclient":   "src.models.clients.system_client:SystemToolClient",
    # ...
}

§4.6 Tool System Architecture

4.6.1 ToolDispatcher

class ToolDispatcher:
    """Tool name → callable registry.
    
    15 default tool instances:
    """
    
    _DEFAULT_INSTANCES = [
        ("BaseTools",           BaseTools),
        ("BaseTools.InputController", BaseTools.InputController),
        ("BrowserTools",        BrowserTools),
        ("MultimodalTools",     MultimodalTools),
        ("DailyTools",          DailyTools),
        ("DailyMonitorTools",   DailyMonitorTools),
        ("DocTools",            DocTools),
        ("MemoryTools",         MemoryTools),
        ("SystemTools",         SystemTools),
        ("UserTools",           UserTools),
        ("UserSecretTools",     UserSecretTools),
        ("PhoneTools",          PhoneTools),
        ("SkillTools",          SkillTools),
        ("SkillWatcherTools",   SkillWatcherTools),
        ("BrokerTools",         BrokerTools),
    ]

4.6.2 execute_skill Pseudo-Tool

def execute_skill_schema() -> dict:
    """OpenAI function schema for execute_skill pseudo-tool.
    
    Framework-level affordance, registered on every ToolDispatcher.
    Recipe / flow units expose this schema via tools.whitelist.
    """
    return {
        "type": "function",
        "function": {
            "name": "execute_skill",
            "parameters": {
                "type": "object",
                "properties": {
                    "skill_id": {"type": "string"},
                    "args": {"type": "object"},
                },
                "required": ["skill_id"],
            },
        },
    }

4.6.3 Tool Activity Mapping

_INSTANCE_ACTIVITY_KEY: dict[str, str] = {
    "BaseTools":         "base",
    "BrowserTools":      "browser",
    "MultimodalTools":   "multimodal",
    "DailyTools":        "daily",
    "DailyMonitorTools": "daily",
    "DocTools":          "doc",
    "SystemTools":       "system",
    "InputController":   "input",
}

Drives the frontend robot-mascot status transitions.

4.6.4 Client Tool Whitelists

Each ToolClient subclass exposes only the tools it can use:

Client	Tool subset
BaseToolClient	Base tools (read / write / execute / find / grep / download)
BrowserToolClient	Browser tools + explore_page
MultimodalToolClient	Vision + operation tools
DocToolClient	Document tools (Word / Excel / PPT / PDF)
DailyToolClient	Schedule / todo / reminder
SystemToolClient	System tools

Schemas come from workspace/tools.json + filtered_schemas() filters per client.

§4.7 Turn Budget Escalator

# max_turns is no longer a hard cap
# When budget is exhausted, asks the manager LLM:
#   - Extend (+N turns)
#   - Abort

# Implementation: _turn_budget_escalator calls MANAGER_MODEL to decide

---

Next: 04 - RAG & Memory →