生产级大模型平台工程实战手册

from pydantic import BaseModel, Fieldclass DocsOutput(BaseModel):    """Result of a grounded documentation lookup."""    answer: str = Field(        description="The answer, grounded entirely in the fetched documentation."    )    sources: list[DocSource] = Field(        default_factory=list,        description="Documentation sources cited; empty only when nothing was fetched.",    )    grounded: bool = Field(        description="True if every claim is backed by the fetched docs; "        "False if the docs did not cover the question.",    )    out_of_scope: bool = Field(        default=False,        description="True if the request fell outside documentation lookup.",    )

{   "answer": "FastAPI's APIRouter lets you group related path operations and mount them on the main app with app.include_router().",   "library_name": "FastAPI",   "version": "0.115",   "sources": [       {           "library_id": "/tiangolo/fastapi",           "topic": "routing"       }   ],   "grounded": true,   "out_of_scope": false}

from google.adk.agents import LlmAgentclass CustomLlmAgent(LlmAgent):    def __init__(self, *args, max_validation_retries: int = 3, **kwargs):        super().__init__(*args, **kwargs)        self._max_validation_retries = max_validation_retries    @property    def max_validation_retries(self) -> int:        """How many times structured-output validation is retried before failing."""        return self._max_validation_retries    async def _run_async_impl(self, ctx):        """Run the agent with retry logic for validation errors.        Both structured-output paths raise pydantic.ValidationError from        inside ADK's _run_async_impl. This wrapper retries on validation        errors up to max_validation_retries times, logging each failure        before retrying.        """        for attempt in range(self.max_validation_retries):           try:               async for event in super()._run_async_impl(ctx):                   yield event               return  # Success           except ValidationError as exc:               is_last_attempt = attempt == self.max_validation_retries - 1               logger.warning(                   "[%s] VALIDATION ERROR (attempt %d/%d): output failed %s validation "                   "(%d error(s)): %s%s",                   self.name,                   attempt + 1,                   self.max_validation_retries,                   getattr(self.output_schema, "__name__", "?"),                   exc.error_count(),                   exc.errors(include_url=False),                   " -- retrying..." if not is_last_attempt else " -- max retries exceeded",               )               metrics.record_hallucination(get_current_user())               if is_last_attempt:                   raise  # Re-raise on last attempt

from google.adk.plugins import ReflectAndRetryToolPluginapp = App(    root_agent=root_agent,    name="app",    plugins=[        ReflectAndRetryToolPlugin(max_retries=3),    ],)

# Docs Agent Prompt -- v1.0.0 (registered first so v2.0.0 below stays active;   # kept available as a rollback target).   register_prompt(       name="docs_agent",       content="""You are a documentation lookup specialist. Your role is to:               - Answer questions about software libraries and frameworks               - Look up APIs, parameters, and usage patterns from their docs               - Ground every answer in the documentation, not from memory               - Say so explicitly when the docs do not cover a question               - Keep answers concise and cite the library/version used""",       version="1.0.0",       tags=["documentation", "lookup", "specialist"]   )   # Docs Agent Prompt -- v2.0.0 (active)   register_prompt(       name="docs_agent",       content="""You are a documentation lookup specialist. Your job is to answer questions about software libraries and frameworks using their current, official documentation -- never from memory.           You have two tools, used in sequence:           1. resolve-library-id -- pass the library or framework name (e.g. "next.js", "fastapi") to get its Context7-compatible library ID. If the name is ambiguous and several libraries match, choose the one that best fits the user's context and state which you picked.           2. get-library-docs -- pass the resolved library ID, plus a topic when the question is about a specific area (e.g. "routing", "middleware", "authentication"), to retrieve the documentation.           Rules:           - ALWAYS resolve and fetch docs before answering any question about a specific API, parameter, version behavior, or usage pattern. Do not rely on training data for library details, which may be outdated.           - Ground every claim in the fetched documentation. When you give a code example or name a parameter, it must come from the docs you retrieved, not from memory.           - If the fetched docs do not cover the question, say so explicitly rather than guessing. Do not fill gaps with plausible-sounding but unverified API details.           - If you cannot resolve the library at all, tell the user the library wasn't found and ask them to confirm the exact name.           - Keep answers concise and cite which library/version the docs came from.           You do not author new documentation, run code, or search the open web. If a request falls outside documentation lookup, say it's out of scope for this agent.           Output format -- return a single JSON object matching the DocsOutput schema:           - answer: your grounded answer, drawn only from the fetched docs.           - library_name: the library the answer is about (e.g. "FastAPI").           - version: the library version the docs reflect, or null if unknown.           - sources: one entry per docs lookup, each with the resolved library_id and the topic (or null).           - grounded: true only if every claim is backed by the fetched docs; false if the docs did not cover the question.           - out_of_scope: true if the request was not a documentation lookup.           Do not include any text outside the JSON object.""",       version="2.0.0",       tags=["documentation", "lookup", "context7", "grounded"]   )

def rollback(self, name: str, version: str) -> str:    """Repoint a prompt's active version to an existing earlier version.    History is never deleted -- rollback only changes which version    get_prompt(name) returns. Returns the version now active. Raises    KeyError if the prompt or target version is unknown.    """    if name not in self._prompts:        raise KeyError(f"Prompt '{name}' not found in registry")    if version not in self._prompts[name]:        raise KeyError(f"Version '{version}' not found for prompt '{name}'")    self._latest_versions[name] = version    return version

def _prompt_provider(name: str):    # Resolve at run time (ADK InstructionProvider) rather than capturing a    # fixed string at construction, so a prompt rollback via the registry    # takes effect on the next request without rebuilding the agents.    return lambda _ctx: get_prompt(name)

def classify_intent(user_input: str) -> Dict:    """Classify user intent based on YAML keyword rules.    Returns a structured decision the calling agent can act on:      - matched:    {"status": "classified", "agent": <name>, "score": <float>}      - no match:   {"status": "unclassified", "agent": None, ...}                    The request is left unclassified rather than defaulting to a                    fixed agent; the caller decides or asks the user.    """    user_input_lower = user_input.lower()    agent_scores = {}    for agent_name, agent_rules in INTENT_RULES.get("agents", {}).items():        keywords = agent_rules.get("keywords", [])        weight = agent_rules.get("weight", 1.0)        score = sum(1 for kw in keywords if kw in user_input_lower) * weight        agent_scores[agent_name] = score    # No confident match: do NOT default to a specific intent. Hand control    # back so the platform can ask the user or fall through to a grounded path.    if not agent_scores or all(score == 0 for score in agent_scores.values()):        return {            "status": "unclassified",            "agent": None,            "reason": "No intent rule matched the request.",            "available_agents": list(INTENT_RULES.get("agents", {}).keys()),            "next_action": "ask_user_about_web_search",        }    best_agent = max(agent_scores, key=agent_scores.get)    return {"status": "classified", "agent": best_agent, "score": agent_scores[best_agent]}

def issue_token(username: str, role: str, ttl_seconds: int = TOKEN_TTL_SECONDS):    """Mint a signed token for an authenticated user."""    now = datetime.now(timezone.utc)    payload = {        "sub": username,        "role": role,        "iat": int(now.timestamp()),        "exp": int((now + timedelta(seconds=ttl_seconds)).timestamp()),    }    token = jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)    return {"token": token, "token_type": "Bearer", "role": role, ...}

developer:  description: Development-only access (no production)  permissions:    tools:      - read_file      - list_directory      - query_database    resources:      - "file://src/*"      - "db://dev/*"      - "db://staging/*"    operations: [SELECT, GET, LIST]  # Explicitly blocked: production databases, deployment, admin operations

from fastmcp.server.middleware import Middleware, MiddlewareContextclass AuthzMiddleware(Middleware):    """Deny tool calls whose caller role isn't permitted by the policy."""    async def on_call_tool(self, context: MiddlewareContext, call_next):        tool_name = context.message.name        auth_header = get_http_headers(include={"authorization"}).get("authorization")        try:            claims = decode_token(token_from_header(auth_header))        except AuthError as exc:            raise ToolError(f"Unauthorized: {exc}")        role = claims.get("role", "")        if not policy_engine.can_access_tool(role, tool_name):            logger.warning("MCP authz DENY role=%r tool=%r", role, tool_name)            raise ToolError(                f"Forbidden: role '{role}' is not authorized to use tool '{tool_name}'"            )        return await call_next(context)

ROLE_TEAM: dict[str, str] = {   "viewer": "shared",   "analyst": "analytics",   "developer": "developer",   "deployer": "devops",   "admin"
本文来源：InfoQ