SnowflakeCortexAgent(
self,
**kwargs = {},
)Create a new Cortex Agent.
Get agent description and configuration.
Update agent configuration.
Delete the agent.
List all agents in the schema.
Create a new Cortex Agent (async).
Get agent description and configuration (async).
Update agent configuration (async).
Delete the agent (async).
List all agents in the schema (async).
Create a new conversation thread.
Update thread configuration.
Delete a conversation thread.
List all conversation threads (global, not agent-specific).
Get thread details and configuration.
Create a new conversation thread (async).
Update thread configuration (async).
Delete a conversation thread (async).
List all conversation threads (global, not agent-specific) (async).
Get thread details and configuration (async).
Snowflake Cortex Agent integration with thread management and feedback.
This class provides access to Snowflake's Cortex Agents REST API, enabling managed orchestration of multiple tools with conversation management.
Setup:
Install langchain-snowflake and configure Snowflake connection.
.. code-block:: bash
pip install -U langchain-snowflake
Key init args:
name: str Name of the Cortex Agent in Snowflake database: str Database containing the agent schema: str Schema containing the agent session: Optional[Session] Active Snowflake session auto_create_threads: bool Whether to automatically create threads (default: True)
Instantiate:
.. code-block:: python
from langchain_snowflake import SnowflakeCortexAgent
agent = SnowflakeCortexAgent( name="sales_assistant", database="sales_db", schema="analytics", session=session )
Direct Usage:
.. code-block:: python
response = agent.invoke("What were Q3 sales trends?")
thread_id = agent.create_thread(metadata={"customer": "ACME"}) response = agent.invoke_with_thread("Show ACME sales", thread_id)
usage = agent.last_usage print(f"Tokens used: {usage['total_tokens']}")
from langchain_snowflake.agents.schemas import FeedbackInput feedback = FeedbackInput( request_id=response["run_id"], positive=True, feedback_message="Good analysis", categories=["accuracy"] ) agent.submit_feedback(feedback)
Async Usage:
.. code-block:: python
response = await agent.ainvoke("What were Q3 sales trends?")
async for chunk in agent.astream("Generate sales report"): print(chunk, end="", flush=True)