create_deep_agent(
    model: str | BaseChatModel | None = None,
    tools: Sequence[...] | None = None,
    *,
    system_prompt: str | SystemMessage | None = None,
    middleware: Sequence[AgentMiddleware] = (),
    subagents: list[...] | None = None,
    skills: list[str] | None = None,
    memory: list[str] | None = None,
    response_format: ResponseFormat | None = None,
    context_schema: type[Any] | None = None,
    ...
) -> CompiledStateGraph

from deepagents import create_deep_agent

agent = create_deep_agent(
    model="openai:gpt-5.5",
    system_prompt="You are an enterprise support assistant.",
)

from deepagents import create_deep_agent
from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    model="gpt-5.5",
    timeout=30,
    max_retries=2,
)

agent = create_deep_agent(
    model=model,
    system_prompt="You are an enterprise support assistant.",
)

agent = create_deep_agent(
    model="openai:gpt-5.5",
    system_prompt="""
You are ACME's enterprise support assistant.
Prioritize policy accuracy, cite internal policy IDs when available,
and escalate instead of guessing.
""",
)

excluded_tools={"execute"}

from pydantic import BaseModel, Field

class SupportDecision(BaseModel):
    priority: str = Field(description="low / medium / high / critical")
    needs_escalation: bool
    reply: str

agent = create_deep_agent(
    model="openai:gpt-5.5",
    response_format=SupportDecision,
)

from langchain.agents.structured_output import ProviderStrategy, ToolStrategy

response_format = ProviderStrategy(SupportDecision)
# 或
response_format = ToolStrategy(SupportDecision)

import os
from collections.abc import Callable
from typing import Literal

from deepagents import (
    GeneralPurposeSubagentProfile,
    HarnessProfile,
    create_deep_agent,
    register_harness_profile,
)
from langchain.agents.middleware import (
    ModelRequest,
    ModelResponse,
    wrap_model_call,
    wrap_tool_call,
)
from langchain.messages import SystemMessage, ToolMessage
from langchain.tools import tool
from langchain.tools.tool_node import ToolCallRequest
from langchain_anthropic import ChatAnthropic
from langchain_openai import ChatOpenAI
from langgraph.types import Command
from pydantic import BaseModel, Field


os.environ["OPENAI_API_KEY"] = "sk-..."
os.environ["ANTHROPIC_API_KEY"] = "sk-ant-..."


TENANT_CONTEXT = {
    "company": "ACME Cloud",
    "support_tier": "enterprise",
    "compliance": ["SOC 2", "ISO 27001"],
    "regions": ["us-east-1", "eu-west-1"],
}


@tool
def search_kb(query: str) -> list[dict]:
    """Search internal knowledge base articles relevant to the user's issue."""
    return [
        {
            "doc_id": "KB-1421",
            "title": "SAML login loops after IdP metadata rotation",
            "summary": "Rotate the SP certificate and re-upload IdP metadata.",
        },
        {
            "doc_id": "KB-1770",
            "title": "API 429 spikes during regional failover",
            "summary": "Enable burst queueing and lower retry concurrency.",
        },
    ]


@tool
def lookup_account_policy(policy_key: Literal["sla", "security", "refund"]) -> dict:
    """Return the enterprise policy currently in force for the requested policy area."""
    policies = {
        "sla": {
            "policy_id": "POL-SLA-009",
            "content": "Critical production incidents require 15-minute first response.",
        },
        "security": {
            "policy_id": "POL-SEC-014",
            "content": "Never disclose tenant-isolated diagnostics in a customer reply.",
        },
        "refund": {
            "policy_id": "POL-BILL-003",
            "content": "Billing credits above 20% must be approved by finance.",
        },
    }
    return policies[policy_key]


@tool
def draft_escalation(
    severity: Literal["high", "critical"],
    summary: str,
    owner_team: Literal["identity", "platform", "billing"],
) -> dict:
    """Draft an internal escalation payload for the on-call team."""
    return {
        "ticket_type": "internal_escalation",
        "severity": severity,
        "owner_team": owner_team,
        "summary": summary,
    }


class SupportDecision(BaseModel):
    """Final structured result expected from the support assistant."""

    issue_type: Literal["how_to", "bug", "billing", "security", "incident"]
    priority: Literal["low", "medium", "high", "critical"]
    needs_escalation: bool
    cited_policy_ids: list[str] = Field(
        description="Policy IDs or KB IDs explicitly referenced in the answer."
    )
    customer_reply: str = Field(
        description="Final customer-facing answer. Clear, professional, and actionable."
    )
    internal_summary: str = Field(
        description="Short internal summary for CRM or handoff."
    )


FAST_MODEL = ChatOpenAI(
    model="gpt-5.5",
    timeout=20,
    max_retries=2,
)

DEEP_MODEL = ChatAnthropic(
    model="claude-opus-4-6",
    timeout=45,
    max_retries=2,
)


@wrap_model_call
def add_enterprise_context(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
    """Append tenant and compliance context without replacing existing prompt layers."""
    extra_blocks = list(request.system_message.content_blocks) + [
        {
            "type": "text",
            "text": (
                "Enterprise context:\n"
                f"- Company: {TENANT_CONTEXT['company']}\n"
                f"- Support tier: {TENANT_CONTEXT['support_tier']}\n"
                f"- Compliance: {', '.join(TENANT_CONTEXT['compliance'])}\n"
                f"- Regions: {', '.join(TENANT_CONTEXT['regions'])}\n"
                "- Never expose internal-only diagnostics or raw tenant data.\n"
                "- If confidence is low, recommend escalation instead of guessing."
            ),
        }
    ]
    return handler(
        request.override(system_message=SystemMessage(content=extra_blocks))
    )


@wrap_model_call
def route_model(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
    """Use a stronger model for incident/security/long-context cases."""
    last_user_text = str(request.messages[-1].content).lower()
    complex_signals = (
        "security",
        "incident",
        "root cause",
        "outage",
        "sev1",
        "critical",
    )
    chosen_model = (
        DEEP_MODEL
        if len(request.messages) > 10
        or any(signal in last_user_text for signal in complex_signals)
        else FAST_MODEL
    )
    return handler(request.override(model=chosen_model))


@wrap_tool_call
def audit_tool_calls(
    request: ToolCallRequest,
    handler: Callable[[ToolCallRequest], ToolMessage | Command],
) -> ToolMessage | Command:
    """Simple audit log for tool usage."""
    print(f"[tool] {request.tool_call['name']} args={request.tool_call['args']}")
    return handler(request)


register_harness_profile(
    "openai",
    HarnessProfile(
        system_prompt_suffix=(
            "Prefer concise customer-facing prose. "
            "Cite policy IDs or KB IDs whenever they materially support the answer."
        ),
        excluded_tools={"execute"},
        tool_description_overrides={
            "task": (
                "Delegate only when the request clearly benefits from isolated "
                "research or multi-step analysis."
            )
        },
        general_purpose_subagent=GeneralPurposeSubagentProfile(
            system_prompt=(
                "You are ACME's research subagent. "
                "Gather evidence, cite policy IDs and KB IDs, and avoid drafting the final customer reply."
            )
        ),
    ),
)

register_harness_profile(
    "anthropic",
    HarnessProfile(
        system_prompt_suffix=(
            "Be methodical. Keep private reasoning internal, but return a concise final answer."
        ),
        excluded_tools={"execute"},
    ),
)

register_harness_profile(
    "openai:gpt-5.5",
    HarnessProfile(
        system_prompt_suffix=(
            "When multiple valid remedies exist, prefer the one with the lowest operational risk."
        )
    ),
)


def build_agent(model_spec):
    """
    model_spec can be either:
    - a provider:model string, e.g. 'openai:gpt-5.5'
    - a preconfigured model instance, e.g. ChatAnthropic(...)
    """
    return create_deep_agent(
        model=model_spec,
        tools=[search_kb, lookup_account_policy, draft_escalation],
        system_prompt=(
            "You are ACME's enterprise support assistant. "
            "Your job is to diagnose the issue, use policies and KB evidence, "
            "draft a safe customer reply, and escalate when risk is high."
        ),
        middleware=[
            add_enterprise_context,
            route_model,
            audit_tool_calls,
        ],
        response_format=SupportDecision,
        name="acme_enterprise_support",
    )


agent = build_agent("openai:gpt-5.5")

result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": (
                    "Our enterprise tenant is seeing repeated SAML login loops after "
                    "rotating IdP metadata. Please explain likely causes, cite any "
                    "relevant KB or policy, and tell me whether this needs escalation."
                ),
            }
        ]
    }
)

print(result["structured_response"].model_dump())


# If you want to pass a model instance instead of a string:
vip_agent = build_agent(
    ChatAnthropic(
        model="claude-opus-4-6",
        timeout=60,
        max_retries=2,
    )
)

result["structured_response"]