Skip to content

symfonic.cli.chat

chat

symfonic chat routing through the public Agent.

HMS is an explicit durable composition, not a property of bare Agent. The HMS route requires Postgres and never constructs SymfonicAgent.

ChatError

Bases: RuntimeError

A chat invocation that cannot proceed, with an adopter-facing message.

ChatOutcome dataclass 

ChatOutcome(text: str, runtime: Runtime, system_prompt_tokens: int | None, memory_entries_used: int, duration_ms: float, native_hms: NativeHmsOutcome | None = None)

What the turn produced, plus which runtime produced it.

ChatRequest dataclass 

ChatRequest(query: str, tenant_id: str, dsn: str | None, model: str, enable_hms: bool, domain: str, retrieval_min_relevance: float = 0.0)

One symfonic chat invocation, decoded off the command line.

build_provider 

build_provider(model: str, query: str) -> Any

Resolve --model to a provider instance.

Raises:

Type Description
ChatError

for an unknown name, or for a provider whose optional extra is not installed — with the pip install line in the message, because a traceback here is a packaging question, not a bug.
Source code in src/symfonic/cli/chat.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def build_provider(model: str, query: str) -> Any:
    """Resolve ``--model`` to a provider instance.

    Raises:
        ChatError: for an unknown name, or for a provider whose optional extra
            is not installed — with the ``pip install`` line in the message,
            because a traceback here is a packaging question, not a bug.
    """
    if model == "mock":
        from symfonic.core.testing import MockModelProvider

        return MockModelProvider(response=f"[mock] Responding to: {query}")

    if model == "anthropic":
        try:
            from symfonic.core.providers import AnthropicProvider  # pragma: no cover
        except ImportError as exc:
            raise ChatError(
                "Anthropic provider requires: pip install symfonic-core[anthropic]"
            ) from exc
        return AnthropicProvider()  # pragma: no cover

    raise ChatError(f"Unknown model provider: {model}. Use 'mock' or 'anthropic'.")

run_chat async 

run_chat(request: ChatRequest, provider: Any) -> ChatOutcome

Execute one turn on whichever runtime :func:select_runtime picks.

Source code in src/symfonic/cli/chat.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
async def run_chat(request: ChatRequest, provider: Any) -> ChatOutcome:
    """Execute one turn on whichever runtime :func:`select_runtime` picks."""
    runtime = select_runtime(request)
    start = time.monotonic()
    if runtime == "native_hms":
        if not request.dsn or not request.dsn.strip():
            raise ChatError("--enable-hms requires --dsn with a PostgreSQL database")
        if request.model == "mock":
            raise ChatError("--enable-hms requires a real model provider, not mock")
        from symfonic.cli.native_hms import run_native_hms_turn

        native = await run_native_hms_turn(
            query=request.query,
            tenant_id=request.tenant_id,
            dsn=request.dsn,
            domain=request.domain,
            provider=provider,
            retrieval_min_relevance=request.retrieval_min_relevance,
        )
        return ChatOutcome(
            text=native.text,
            runtime=runtime,
            system_prompt_tokens=None,
            memory_entries_used=native.recalled_records_count,
            duration_ms=(time.monotonic() - start) * 1000,
            native_hms=native,
        )
    if request.domain != GENERIC_DOMAIN or request.dsn:
        text, system_tokens, memory_used = await _run_presented(request, provider)
    else:
        text, system_tokens, memory_used = await _run_facade(request, provider)
    return ChatOutcome(
        text=text,
        runtime=runtime,
        system_prompt_tokens=system_tokens,
        memory_entries_used=memory_used,
        duration_ms=(time.monotonic() - start) * 1000,
    )

select_runtime 

select_runtime(request: ChatRequest) -> Runtime

Return the runtime this invocation needs.

HMS opts into the durable native composition. Without HMS, named domains and DSNs retain the presented public-Agent route.

Source code in src/symfonic/cli/chat.py 
64
65
66
67
68
69
70
71
72
def select_runtime(request: ChatRequest) -> Runtime:
    """Return the runtime this invocation needs.

    HMS opts into the durable native composition. Without HMS, named domains
    and DSNs retain the presented public-Agent route.
    """
    if request.enable_hms:
        return "native_hms"
    return "facade"