在Streamlit in Snowflake中接入自定义Snowflake Cortex Agent遇阻求助

当前可通过Semantic View连接并获取有效查询结果，但尝试接入自定义Cortex Agent以获取自然语言响应时遇到问题，相关代码如下：

配置代码 #

AGENT_NAME = "A.B.MY_CUSTOM_AGENT"

# Choose exactly ONE way to point Analyst at your semantics:
USE_SEMANTIC_VIEW = True
SEMANTIC_VIEW = "A.B.SEMANTIC_VIEW"  

API_ENDPOINT = "/api/v2/cortex/agent:run"

Agent调用代码 #

from typing import List, Dict, Any
import json
import streamlit as st

def call_cortex(conversation_history: List[Dict[str, Any]]) -> Dict[str, Any]:
    tools, tool_resources = build_tools_payload()

    payload = {
        "agent": AGENT_NAME,
        "model": MODEL_NAME,
        "messages": conversation_history, # Pass the full history
        "tool_choice": {"type": "auto"},
        "tools": tools,
        "tool_resources": tool_resources,
    }

    resp = _snowflake.send_snow_api_request(
        "POST",
        API_ENDPOINT,
        {},  # headers
        {},  # params
        payload,
        None,
        API_TIMEOUT_MS,
    )

    status = resp.get("status")
    if status != 200:
        st.error(f"❌ HTTP {status}: {resp.get('reason', '(no reason)')}")
        st.code(resp.get("content", "")[:2000])
        return {"error": f"HTTP {status}"}

    raw = resp.get("content", "")
    if isinstance(raw, str) and raw.strip().startswith("event:"):
        events = parse_sse(raw)
        text = assemble_text_from_deltas(events)
        sqls = extract_sql_snippets(events)
        return {"mode": "sse", "text": text, "sqls": sqls, "events": events}

    try:
        content = json.loads(raw)
    except json.JSONDecodeError:
        st.error("❌ Response is neither JSON nor SSE.")
        st.code(raw[:2000])
        return {"error": "non-json-non-sse"}

    if isinstance(content, list):
        events = []
        for item in content:
            if isinstance(item, dict) and "event" in item:
                events.append(item)
        text = assemble_text_from_deltas(events) if events else ""
        sqls = extract_sql_snippets(events) if events else []
        return {"mode": "json_deltas", "text": text, "sqls": sqls, "events": events}

    if isinstance(content, dict):
        msgs = content.get("messages") or []
        text = ""
        if msgs:
            last = msgs[-1]
            for part in (last.get("content") or []):
                if isinstance(part, dict) and part.get("type") == "text":
                    text += part.get("text", "")
        return {"mode": "json", "text": text, "sqls": [], "raw": content}

    return {"error": "unexpected-shape", "raw": raw}

排查关键点 #

• 配置冲突检查：接入自定义Agent时需禁用Semantic View配置，将USE_SEMANTIC_VIEW设为False，避免两种语义指向方式冲突。
• 权限与端点验证：
  • 确认当前SiS应用的角色拥有自定义AgentA.B.MY_CUSTOM_AGENT的USAGE权限。
  • 验证/api/v2/cortex/agent:run端点是否正确，可通过Snowflake官方文档确认最新格式。
• Payload参数优化：
  • 若自定义Agent已绑定模型，移除payload中的model字段，避免参数冲突。
  • 若Agent已预设工具集，将tools设为[]、tool_resources设为{}，测试是否因手动传入工具导致冲突。
• 响应适配检查：自定义Agent的响应格式可能与Semantic View不同，验证parse_sse、assemble_text_from_deltas等解析函数是否适配Agent输出结构。
• Agent可用性测试：在Snowflake Worksheet中执行CALL SYSTEM_RUN_CORTEX_AGENT('A.B.MY_CUSTOM_AGENT', '测试查询')，确认Agent本身能正常返回自然语言响应。

内容的提问来源于stack exchange，提问作者Samir112