AI Agent 开发学习与面试指南：10-综合项目：企业知识与行动助手

Web UI
  -> FastAPI / Auth / Rate Limit
      -> Run Service -> PostgreSQL Checkpoint
      -> Agent Graph
          -> Intent Router
          -> RAG Service -> BM25 + Vector -> Reranker
          -> Tool Gateway
              -> get_ticket_status
              -> create_ticket_draft
              -> submit_ticket
          -> Approval Node
          -> Human Handoff
      -> Redis Queue / Cache
      -> Trace + Metrics + Audit

enterprise-support-agent/
├─ app/
│  ├─ api/
│  │  ├─ runs.py
│  │  └─ approvals.py
│  ├─ agent/
│  │  ├─ graph.py
│  │  ├─ state.py
│  │  ├─ nodes.py
│  │  └─ policies.py
│  ├─ rag/
│  │  ├─ ingest.py
│  │  ├─ chunking.py
│  │  ├─ retrieval.py
│  │  └─ citations.py
│  ├─ tools/
│  │  ├─ registry.py
│  │  ├─ tickets.py
│  │  └─ executor.py
│  ├─ eval/
│  │  ├─ datasets/
│  │  ├─ evaluators.py
│  │  └─ regression_gate.py
│  ├─ infra/
│  │  ├─ model_gateway.py
│  │  ├─ database.py
│  │  ├─ cache.py
│  │  └─ tracing.py
│  └─ main.py
├─ tests/
│  ├─ unit/
│  ├─ integration/
│  ├─ eval/
│  └─ security/
├─ migrations/
├─ docs/
│  ├─ architecture.md
│  ├─ threat-model.md
│  └─ failure-review.md
├─ Dockerfile
├─ compose.yaml
├─ pyproject.toml
└─ README.md

create table agent_runs (
    run_id uuid primary key,
    user_id text not null,
    tenant_id text not null,
    status text not null,
    graph_version text not null,
    prompt_version text not null,
    model_config jsonb not null,
    created_at timestamptz not null,
    updated_at timestamptz not null,
    deadline_at timestamptz not null
);

create table tool_executions (
    execution_id uuid primary key,
    run_id uuid not null references agent_runs(run_id),
    tool_name text not null,
    tool_version text not null,
    idempotency_key text not null,
    risk_level text not null,
    status text not null,
    arguments_hash text not null,
    result_reference text,
    created_at timestamptz not null,
    unique(tool_name, idempotency_key)
);

create table approvals (
    approval_id uuid primary key,
    run_id uuid not null references agent_runs(run_id),
    action text not null,
    proposed_arguments jsonb not null,
    status text not null,
    decided_by text,
    decided_at timestamptz,
    expires_at timestamptz not null
);

from typing import Literal, TypedDict


class SupportAgentState(TypedDict, total=False):
    run_id: str
    user_id: str
    tenant_id: str
    query: str
    intent: Literal["knowledge", "ticket_query", "ticket_create", "human"]
    evidence: list[dict]
    ticket_id: str | None
    draft: dict | None
    approval_id: str | None
    approved: bool | None
    tool_trace: list[dict]
    step_count: int
    final_answer: str
    error: dict | None

START
 -> normalize_input
 -> route_intent
     knowledge -> retrieve -> rerank -> answer_with_citations -> END
     ticket_query -> authorize -> get_ticket -> answer -> END
     ticket_create -> collect_fields -> create_draft -> approval
         approved -> submit_ticket -> answer -> END
         rejected -> cancel_draft -> END
     human -> handoff -> END

class GetTicketInput(BaseModel):
    ticket_id: str = Field(pattern=r"^T-[0-9]{6}$")


async def get_ticket_status(args: GetTicketInput, auth_context) -> dict:
    ticket = await repository.get(args.ticket_id)
    if ticket.tenant_id != auth_context.tenant_id:
        raise PermissionDenied()
    if ticket.user_id != auth_context.user_id and not auth_context.can("ticket:read:any"):
        raise PermissionDenied()
    return ticket.to_safe_dict()

Query Rewrite
 -> tenant/ACL/version filters
 -> Dense Top 30 + BM25 Top 30
 -> RRF merge
 -> Rerank Top 6
 -> diversify by document
 -> context budget

run_id=...
route_intent: ticket_create
collect_fields: category=technical, missing=[]
create_draft: success, draft_id=D-12
approval: approved by u-100
submit_ticket: timeout after 3s
idempotency_lookup: found T-123456
final: success, no duplicate write