人机协作工具

from a13n_harness import HarnessBuilder
from a13n_harness.capabilities import UserInteractionCapability

executable = HarnessBuilder().build(
    agent_spec,
    output_type=str,
    model=model,
    capabilities=(UserInteractionCapability(),),
)

{
  "questions": [
    {
      "question": "Which scope should be used?",
      "header": "Scope",
      "options": [
        {"label": "Focused", "description": "Change only the selected component."},
        {"label": "Broad", "description": "Apply the change across the repository."}
      ],
      "multiSelect": false
    },
    {
      "question": "Which outputs do you need?",
      "header": "Outputs",
      "options": [
        {"label": "Code", "description": "Include the implementation."},
        {"label": "Docs", "description": "Include the user guide."}
      ],
      "multiSelect": true
    }
  ]
}

{
  "answers": {
    "Which scope should be used?": "Focused",
    "Which outputs do you need?": ["Code", "Docs"]
  }
}

{
  "answers": {
    "Which scope should be used?": "Only update the Python integration.",
    "Which outputs do you need?": "A runnable example is enough."
  }
}

{
  "response": "Keep the implementation focused and include a short guide."
}

export interface UserQuestionOption {
  label: string;
  description: string;
}

export interface UserQuestion {
  question: string;
  header: string;
  options: UserQuestionOption[];
  multiSelect?: boolean;
}

export interface AskUserQuestionRequest {
  questions: UserQuestion[];
}

export interface UserQuestionAnswers {
  answers?: Record<string, string | string[]>;
  response?: string | null;
}

from a13n_harness.capabilities import (
    AskUserQuestionRequest,
    UserQuestion,
    UserQuestionAnswers,
    UserQuestionOption,
)

request = AskUserQuestionRequest.model_validate_json(request_json)
answer = UserQuestionAnswers.model_validate_json(answer_json)
request_payload = request.model_dump(mode="json", by_alias=True)
answer_payload = answer.model_dump(mode="json", exclude_none=True)

from a13n_harness import DeferredToolResume, RunBindings
from a13n_harness.capabilities import AskUserQuestionRequest


async def answer_question(executable):
    first = await executable.run("Clarify the scope", bindings=RunBindings.embedded())
    if first.status != "suspended" or first.state is None or first.deferred is None:
        raise RuntimeError("Expected a pending question")
    requests = first.deferred
    if len(requests.calls) != 1 or requests.approvals:
        raise RuntimeError("This example expects one call and no approvals")
    call = requests.calls[0]
    if call.tool_name != "ask_user_question":
        raise RuntimeError("Expected the built-in question tool")
    question_request = AskUserQuestionRequest.model_validate(call.args_as_dict())
    answer = {
        "answers": {
            question.question: question.options[0].label
            for question in question_request.questions
        }
    }
    results = requests.build_results(calls={call.tool_call_id: answer})
    second = await executable.run(
        bindings=RunBindings.embedded(),
        previous_state=first.state,
        deferred_resume=DeferredToolResume(requests, results),
    )
    return second.output_or_raise()

{
  "approvals": {},
  "calls": {
    "question-1": {
      "status": "returned",
      "value": {
        "answers": {
          "Which scope should be used?": "Focused",
          "Which outputs do you need?": ["Code", "Docs"]
        }
      }
    }
  }
}