docs(examples): add conversation/identity_config/custom_tools/streaming_deltas to match Go parity

Task: 1789906941

Author: moonyue-w

examples/forward/__main__.py

@@ -4,19 +4,25 @@
 from qca import Forward
 
 from .batch import run as batch
+from .conversation import run as conversation
+from .identity_config import run as identity_config
 from .memory import run as memory
 from .models import run as models
 from .resources import run as resources
 from .schedule import run as schedule
 from .session import run as session
+from .streaming_deltas import run as streaming_deltas
 
 SCENARIOS = {
     "models": models,
     "session": session,
+    "conversation": conversation,
     "resources": resources,
+    "identity_config": identity_config,
     "memory": memory,
     "schedule": schedule,
     "batch": batch,
+    "streaming_deltas": streaming_deltas,
 }
 
 

examples/forward/conversation.py

@@ -0,0 +1,70 @@
+"""复用同一个 Session 进行多轮对话，并分页读取会话历史。
+
+运行：python -m examples.forward.conversation
+"""
+
+from __future__ import annotations
+
+from examples.common.live import Run, choose_model, marker, name, run_cli
+from qca import Forward
+
+from ._cleanup import finish_session
+
+
+def run(client: Forward, context: Run) -> None:
+    environment = client.environments.create(name=name("env"), config={"type": "cloud"})
+    environment_id = context.track("environment", environment.id, lambda: client.environments.archive(environment.id))
+
+    identity = client.identities.create(external_id=name("identity"), name="SDK 示例用户")
+    identity_id = context.track("identity", identity.id, lambda: client.identities.delete(identity.id))
+
+    model = choose_model(client.models.list(), context.config.model)
+    context.output("selected_model", model)
+    template = client.templates.create(
+        name=name("template"),
+        environment_id=environment_id,
+        model=model,
+        system="你是一个 SDK 示例助手。必要时调用工具，只使用可实际读取的数据回答问题。",
+        tools=[{"type": "agent_toolset_20260401"}],
+    )
+    template_id = context.track("template", template.id, lambda: client.templates.archive(template.id))
+
+    session = client.sessions.create(identity_id=identity_id, template_id=template_id)
+    session_id = context.track("session", session.id, lambda: finish_session(client, context, session.id))
+
+    def ask(prompt: str) -> None:
+        context.output("user", prompt)
+        sent = client.sessions.events.send(
+            session_id,
+            events=[{"type": "user.message", "content": [{"type": "text", "text": prompt}]}],
+            extra_headers={"Idempotency-Key": name("event")},
+        )
+        if not sent.data or not sent.data[0].id:
+            raise RuntimeError("Send returned no user event")
+        with client.sessions.events.stream(
+            session_id,
+            extra_headers={"Last-Event-ID": sent.data[0].id},
+            timeout=context.remaining(),
+        ) as stream:
+            for event in stream:
+                context.remaining()
+                if event.type == "agent.message":
+                    context.output("assistant", event.to_json())
+                elif event.type in ("session.error", "session.status_terminated"):
+                    raise RuntimeError(f"Session stopped: {event.type}")
+                elif event.type == "session.status_idle":
+                    break
+
+    # 同一个 Session 支持多轮：服务端在 session_id 下保留完整历史，无需客户端携带上文。
+    code = "project-" + marker()
+    ask(f"这次项目代号是 {code}。请在本次对话中记住它，不要使用工具或写入记忆库。现在只回复：已记住。")
+    ask("只根据本次会话上文，告诉我刚才约定的项目代号。只回复代号，不要使用工具。")
+
+    # 分页读取已有的用户消息和助手回复，重建对话文字记录。
+    for event in client.sessions.events.list(session_id, order="asc", limit=100):
+        if event.type in ("user.message", "agent.message"):
+            context.output(f"history.{event.type}", event.to_json())
+
+
+if __name__ == "__main__":
+    run_cli("forward", Forward, {"conversation": run})

examples/forward/identity_config.py

@@ -0,0 +1,80 @@
+"""两个 Identity 共用一个 Template，写入各自的个性化配置，再读取生效配置并在会话中验证。
+
+运行：python -m examples.forward.identity_config
+"""
+
+from __future__ import annotations
+
+from examples.common.live import Run, choose_model, marker, name, run_cli
+from qca import Forward
+
+from ._cleanup import finish_session
+
+
+def run(client: Forward, context: Run) -> None:
+    environment = client.environments.create(name=name("env"), config={"type": "cloud"})
+    environment_id = context.track("environment", environment.id, lambda: client.environments.archive(environment.id))
+
+    model = choose_model(client.models.list(), context.config.model)
+    context.output("selected_model", model)
+    shared, baseline = marker(), marker()
+    template = client.templates.create(
+        name=name("template"),
+        environment_id=environment_id,
+        model=model,
+        system="你是一个 SDK 示例助手。必要时调用工具，只使用可实际读取的数据回答问题。",
+        tools=[{"type": "agent_toolset_20260401"}],
+        environment_variables={"SDK_SHARED_VALUE": shared, "SDK_PERSONAL_VALUE": baseline},
+    )
+    template_id = context.track("template", template.id, lambda: client.templates.archive(template.id))
+
+    # 先创建两个 Identity 并写入个性化配置：SDK_SHARED_VALUE 继承模板默认，SDK_PERSONAL_VALUE 被各自覆盖。
+    identities: list[str] = []
+    personal: list[str] = []
+    for index in range(2):
+        identity = client.identities.create(external_id=name("identity"), name=f"SDK 示例用户 {index + 1}")
+        identity_id = context.track("identity", identity.id, lambda ref=identity.id: client.identities.delete(ref))
+        identities.append(identity_id)
+        value = marker()
+        personal.append(value)
+        client.identities.configs.upsert(
+            template_id,
+            identity_id=identity_id,
+            identity_config={"environment_variables": {"SDK_PERSONAL_VALUE": {"op": "set", "value": value}}},
+        )
+        effective = client.identities.configs.retrieve_effective(template_id, identity_id=identity_id)
+        context.output(
+            f"identity_{index + 1}_effective_env",
+            effective.session.environment_variables if effective.session else None,
+        )
+
+    # 每个 Identity 各起一个会话，读取自己实际生效的环境变量。
+    for index, identity_id in enumerate(identities):
+        session = client.sessions.create(identity_id=identity_id, template_id=template_id)
+        session_id = context.track("session", session.id, lambda ref=session.id: finish_session(client, context, ref))
+        prompt = "请使用工具读取 SDK_SHARED_VALUE 和 SDK_PERSONAL_VALUE 两个环境变量，只返回这两个变量的实际值。"
+        context.output(f"identity_{index + 1}_user", prompt)
+        sent = client.sessions.events.send(
+            session_id,
+            events=[{"type": "user.message", "content": [{"type": "text", "text": prompt}]}],
+            extra_headers={"Idempotency-Key": name("event")},
+        )
+        if not sent.data or not sent.data[0].id:
+            raise RuntimeError("Send returned no user event")
+        with client.sessions.events.stream(
+            session_id,
+            extra_headers={"Last-Event-ID": sent.data[0].id},
+            timeout=context.remaining(),
+        ) as stream:
+            for event in stream:
+                context.remaining()
+                if event.type == "agent.message":
+                    context.output(f"identity_{index + 1}_assistant", event.to_json())
+                elif event.type in ("session.error", "session.status_terminated"):
+                    raise RuntimeError(f"Session stopped: {event.type}")
+                elif event.type == "session.status_idle":
+                    break
+
+
+if __name__ == "__main__":
+    run_cli("forward", Forward, {"identity_config": run})

examples/forward/streaming_deltas.py

@@ -0,0 +1,84 @@
+"""订阅 agent.message 文本增量，逐步更新消息预览，再用最终完整消息替换预览。
+
+运行：python -m examples.forward.streaming_deltas
+"""
+
+from __future__ import annotations
+
+from examples.common.live import Run, choose_model, marker, name, run_cli
+from qca import Forward
+
+from ._cleanup import finish_session
+
+
+def _delta_text(event: object) -> str:
+    """从一条文本增量事件中取出片段文本；不是文本增量时返回空串。"""
+    delta = event.to_dict().get("delta") or {}
+    if not isinstance(delta, dict) or delta.get("type") != "content_delta":
+        return ""
+    content = delta.get("content") or {}
+    if not isinstance(content, dict) or content.get("type") != "text":
+        return ""
+    return content.get("text") or ""
+
+
+def run(client: Forward, context: Run) -> None:
+    environment = client.environments.create(name=name("env"), config={"type": "cloud"})
+    environment_id = context.track("environment", environment.id, lambda: client.environments.archive(environment.id))
+
+    identity = client.identities.create(external_id=name("identity"), name="SDK 示例用户")
+    identity_id = context.track("identity", identity.id, lambda: client.identities.delete(identity.id))
+
+    model = choose_model(client.models.list(), context.config.model)
+    context.output("selected_model", model)
+    template = client.templates.create(
+        name=name("template"),
+        environment_id=environment_id,
+        model=model,
+        system="你是一个 SDK 示例助手。必要时调用工具，只使用可实际读取的数据回答问题。",
+        tools=[{"type": "agent_toolset_20260401"}],
+    )
+    template_id = context.track("template", template.id, lambda: client.templates.archive(template.id))
+
+    session = client.sessions.create(identity_id=identity_id, template_id=template_id)
+    session_id = context.track("session", session.id, lambda: finish_session(client, context, session.id))
+
+    marker_value = marker()
+    prompt = "请分三句话解释为什么多轮对话要复用 Session ID，最后原样附上：" + marker_value
+    # 预览增量不写入历史：先订阅、开启 agent.message 增量，再发送消息。
+    with client.sessions.events.stream(
+        session_id,
+        event_deltas=["agent.message"],
+        timeout=context.remaining(),
+    ) as stream:
+        context.output("user", prompt)
+        sent = client.sessions.events.send(
+            session_id,
+            events=[{"type": "user.message", "content": [{"type": "text", "text": prompt}]}],
+            extra_headers={"Idempotency-Key": name("event")},
+        )
+        if not sent.data or not sent.data[0].id:
+            raise RuntimeError("Send returned no user event")
+
+        previews: dict[str, str] = {}
+        deltas = 0
+        for event in stream:
+            context.remaining()
+            if event.type == "event_delta":
+                text = _delta_text(event)
+                if text and event.event_id:
+                    deltas += 1
+                    # 同一条消息的多个片段按 event_id 累加，逐步刷新该消息的预览。
+                    previews[event.event_id] = previews.get(event.event_id, "") + text
+                    context.output("preview", f"{event.event_id}: {previews[event.event_id]}")
+            elif event.type == "agent.message":
+                context.output("assistant", event.to_json())
+            elif event.type in ("session.error", "session.status_terminated"):
+                raise RuntimeError(f"Session stopped: {event.type}")
+            elif event.type == "session.status_idle":
+                break
+        context.output("deltas", deltas)
+
+
+if __name__ == "__main__":
+    run_cli("forward", Forward, {"streaming_deltas": run})

examples/managed/__main__.py

@@ -3,20 +3,26 @@
 from examples.common.live import run_cli
 from qca import Managed
 
+from .conversation import run as conversation
+from .custom_tools import run as custom_tools
 from .deployment import run as deployment
 from .dream import run as dream
 from .memory import run as memory
 from .models import run as models
 from .resources import run as resources
 from .session import run as session
+from .streaming_deltas import run as streaming_deltas
 
 SCENARIOS = {
     "models": models,
     "session": session,
+    "conversation": conversation,
     "resources": resources,
+    "custom_tools": custom_tools,
     "memory": memory,
     "deployment": deployment,
     "dream": dream,
+    "streaming_deltas": streaming_deltas,
 }
 
 

examples/managed/conversation.py

@@ -0,0 +1,66 @@
+"""复用同一个 Session 进行多轮对话，并分页读取会话历史。
+
+运行：python -m examples.managed.conversation
+"""
+
+from __future__ import annotations
+
+from examples.common.live import Run, choose_model, marker, name, run_cli
+from qca import Managed
+
+from ._cleanup import finish_session
+
+
+def run(client: Managed, context: Run) -> None:
+    environment = client.environments.create(name=name("env"), config={"type": "cloud"})
+    environment_id = context.track("environment", environment.id, lambda: client.environments.archive(environment.id))
+
+    model = choose_model(client.models.list(), context.config.model)
+    context.output("selected_model", model)
+    agent = client.agents.create(
+        name=name("agent"),
+        model={"id": model},
+        system="你是一个 SDK 示例助手。必要时调用工具，只使用可实际读取的数据回答问题。",
+        tools=[{"type": "agent_toolset_20260401"}],
+    )
+    agent_id = context.track("agent", agent.id, lambda: client.agents.archive(agent.id))
+
+    session = client.sessions.create(environment_id=environment_id, agent=agent_id)
+    session_id = context.track("session", session.id, lambda: finish_session(client, context, session.id))
+
+    def ask(prompt: str) -> None:
+        context.output("user", prompt)
+        sent = client.sessions.events.send(
+            session_id,
+            events=[{"type": "user.message", "content": [{"type": "text", "text": prompt}]}],
+            extra_headers={"Idempotency-Key": name("event")},
+        )
+        if not sent.data or not sent.data[0].id:
+            raise RuntimeError("Send returned no user event")
+        with client.sessions.events.stream(
+            session_id,
+            extra_headers={"Last-Event-ID": sent.data[0].id},
+            timeout=context.remaining(),
+        ) as stream:
+            for event in stream:
+                context.remaining()
+                if event.type == "agent.message":
+                    context.output("assistant", event.to_json())
+                elif event.type in ("session.error", "session.status_terminated"):
+                    raise RuntimeError(f"Session stopped: {event.type}")
+                elif event.type == "session.status_idle":
+                    break
+
+    # 同一个 Session 支持多轮：服务端在 session_id 下保留完整历史，无需客户端携带上文。
+    code = "project-" + marker()
+    ask(f"这次项目代号是 {code}。请在本次对话中记住它，不要使用工具或写入记忆库。现在只回复：已记住。")
+    ask("只根据本次会话上文，告诉我刚才约定的项目代号。只回复代号，不要使用工具。")
+
+    # 分页读取已有的用户消息和助手回复，重建对话文字记录。
+    for event in client.sessions.events.list(session_id, order="asc", limit=100):
+        if event.type in ("user.message", "agent.message"):
+            context.output(f"history.{event.type}", event.to_json())
+
+
+if __name__ == "__main__":
+    run_cli("managed", Managed, {"conversation": run})