{
  "function": {
    "name": "create_payment_link",
    "parameters": {
      "amount": 750,
      "customer": "cust_128934ddasf9",
      "product": "prod_8675309",
      "price": "prc_09874329fds",
      "quantity": 1,
      "memo": "嘿Jeff - 請參見下方用于二月AI Tinkerers聚會的支付鏈接"
    }
  }
}

# LLM接收自然語言并返回一個結構化對象
nextStep = await llm.determineNextStep(
"""
  為Jeff創(chuàng)建一個750美元的支付鏈接
  用于贊助二月的AI Tinkerers聚會
  """
  )

# 根據(jù)其功能處理結構化輸出
if nextStep.function == 'create_payment_link':
    stripe.paymentlinks.create(nextStep.parameters)
    return# 或者你想要的任何內容，請參見下文
elif nextStep.function == 'something_else':
    # ... 更多情況
    pass
else:  # 模型未調用我們知道的工具
    # 做其他事情
    pass

agent = Agent(
  role="...",
  goal="...",
  persnotallow="...",
  tools=[tool1, tool2, tool3]
)

task = Task(
  instructinotallow="...",
  expected_output=OutputModel
)

result = agent.run(task)

function DetermineNextStep(thread: string) -> DoneForNow | ListGitTags | DeployBackend | DeployFrontend | RequestMoreInformation {
  prompt #"
    {{ _.role("system") }}
    
    你是一個管理前端和后端系統(tǒng)部署的有用助手。
    你勤奮地確保通過遵循最佳實踐和正確的部署程序來實現(xiàn)安全和成功的部署。
    
    在部署任何系統(tǒng)之前，你應該檢查：
    - 部署環(huán)境（暫存 vs 生產(chǎn)）
    - 要部署的正確標簽/版本
    - 當前系統(tǒng)狀態(tài)
    
    你可以使用諸如 deploy_backend、deploy_frontend 和 check_deployment_status
    等工具來管理部署。對于敏感部署，使用 request_approval 來獲得
    人工驗證。
    
    始終思考首先該做什么，比如：
    - 檢查當前部署狀態(tài)
    - 驗證部署標簽是否存在
    - 如有需要，請求批準
    - 在生產(chǎn)之前先部署到暫存環(huán)境
    - 監(jiān)控部署進度
    
    {{ _.role("user") }}

    {{ thread }}
    
    下一步應該是什么？
  "#
}

function DetermineNextStep(thread: string) -> DoneForNow | ListGitTags | DeployBackend | DeployFrontend | RequestMoreInformation {

[
  {
    "role":"system",
    "content":"You are a helpful assistant..."
},
{
    "role":"user",
    "content":"Can you deploy the backend?"
},
{
    "role":"assistant",
    "content":null,
    "tool_calls":[
      {
        "id":"1",
        "name":"list_git_tags",
        "arguments":"{}"
      }
    ]
},
{
    "role":"tool",
    "name":"list_git_tags",
    "content":"{\"tags\": [{\"name\": \"v1.2.3\", \"commit\": \"abc123\", \"date\": \"2024-03-15T10:00:00Z\"}, {\"name\": \"v1.2.2\", \"commit\": \"def456\", \"date\": \"2024-03-14T15:30:00Z\"}, {\"name\": \"v1.2.1\", \"commit\": \"abe033d\", \"date\": \"2024-03-13T09:15:00Z\"}]}",
    "tool_call_id":"1"
}
]

[
{
    "role":"system",
    "content":"You are a helpful assistant..."
},
{
    "role":"user",
    "content":|
            Here's everything that happened so far:
        
        <slack_message>
            From:@alex
            Channel:#deployments
            Text:Canyoudeploythebackend?
        </slack_message>
        
        <list_git_tags>
            intent:"list_git_tags"
        </list_git_tags>
        
        <list_git_tags_result>
            tags:
              -name:"v1.2.3"
                commit:"abc123"
                date:"2024-03-15T10:00:00Z"
              -name:"v1.2.2"
                commit:"def456"
                date:"2024-03-14T15:30:00Z"
              -name:"v1.2.1"
                commit:"ghi789"
                date:"2024-03-13T09:15:00Z"
        </list_git_tags_result>
        
        what'sthenextstep?
    }
]

class Thread:
  events: List[Event]

class Event:
# 可以只用字符串，也可以顯式定義 - 由你決定
  type: Literal["list_git_tags", "deploy_backend", "deploy_frontend", "request_more_information", "done_for_now", "list_git_tags_result", "deploy_backend_result", "deploy_frontend_result", "request_more_information_result", "done_for_now_result", "error"]
  data: ListGitTags | DeployBackend | DeployFrontend | RequestMoreInformation |  
        ListGitTagsResult | DeployBackendResult | DeployFrontendResult | RequestMoreInformationResult | string

def event_to_prompt(event: Event) -> str:
    data = event.data if isinstance(event.data, str) \
           else stringifyToYaml(event.data)

    returnf"<{event.type}>\n{data}\n</{event.type}>"


def thread_to_prompt(thread: Thread) -> str:
return'\n\n'.join(event_to_prompt(event) for event in thread.events)

<slack_message>
    From: @alex
    Channel: #deployments
    Text: Can you deploy the latest backend to production?
</slack_message>

<slack_message>
    From: @alex
    Channel: #deployments
    Text: Can you deploy the latest backend to production?
    Thread: []
</slack_message>

<list_git_tags>
    intent: "list_git_tags"
</list_git_tags>

<list_git_tags_result>
    tags:
      - name: "v1.2.3"
        commit: "abc123"
        date: "2024-03-15T10:00:00Z"
      - name: "v1.2.2"
        commit: "def456"
        date: "2024-03-14T15:30:00Z"
      - name: "v1.2.1"
        commit: "ghi789"
        date: "2024-03-13T09:15:00Z"
</list_git_tags_result>

<slack_message>
    From: @alex
    Channel: #deployments
    Text: Can you deploy the latest backend to production?
    Thread: []
</slack_message>

<deploy_backend>
    intent: "deploy_backend"
    tag: "v1.2.3"
    environment: "production"
</deploy_backend>

<error>
    error running deploy_backend: Failed to connect to deployment service
</error>

<request_more_information>
    intent: "request_more_information_from_human"
    question: "I had trouble connecting to the deployment service, can you provide more details and/or check on the status of the service?"
</request_more_information>

<human_response>
    data:
      response: "I'm not sure what's going on, can you check on the status of the latest workflow?"
</human_response>

nextStep = await determine_next_step(thread_to_prompt(thread))

{
  "intent": "get_workflow_status",
  "workflow_name": "tag_push_prod.yaml",
}

class Issue:
  title: str
  description: str
  team_id: str
  assignee_id: str

class CreateIssue:
  intent: "create_issue"
  issue: Issue

class SearchIssues:
  intent: "search_issues"
  query: str
  what_youre_looking_for: str

if nextStep.intent == 'create_payment_link':
    stripe.paymentlinks.create(nextStep.parameters)
    return # 或者你想要的任何內容，請參見下文
elif nextStep.intent == 'wait_for_a_while': 
    # 做一些單子操作，我不知道
else: #... 模型沒有調用我們知道的工具
    # 做其他事情

class Options:
  urgency: Literal["low", "medium", "high"]
  format: Literal["free_text", "yes_no", "multiple_choice"]
  choices: List[str]

# 用于人機交互的工具定義
class RequestHumanInput:
  intent: "request_human_input"
  question: str
  context: str
  options: Options

# 在代理循環(huán)中的示例用法
if nextStep.intent == 'request_human_input':
  thread.events.append({
    type: 'human_input_requested',
    data: nextStep
  })
  thread_id = await save_state(thread)
await notify_human(nextStep, thread_id)
return# 中斷循環(huán)并等待帶有線程ID的響應
else:
# ... 其他情況

@app.post('/webhook')
def webhook(req: Request):
  thread_id = req.body.threadId
  thread = await load_state(thread_id)
  thread.events.push({
    type: 'response_from_human',
    data: req.body
  })
# ... 為簡潔起見，你可能不希望在這里阻塞 web worker
  next_step = await determine_next_step(thread_to_prompt(thread))
  thread.events.append(next_step)
  result = await handle_next_step(thread, next_step)
# todo - 循環(huán)或中斷或任何你想要的

return {"status": "ok"}

(為簡潔起見省略)

<slack_message>
    From: @alex
    Channel: #deployments
    Text: Can you deploy backend v1.2.3 to production?
    Thread: []
</slack_message>

<request_human_input>
    intent: "request_human_input"
    question: "Would you like to proceed with deploying v1.2.3 to production?"
    context: "This is a production deployment that will affect live users."
    options: {
        urgency: "high"
        format: "yes_no"
    }
</request_human_input>

<human_response>
    response: "yes please proceed"
    approved: true
    timestamp: "2024-03-15T10:30:00Z"
    user: "alex@company.com"
</human_response>

<deploy_backend>
    intent: "deploy_backend"
    tag: "v1.2.3"
    environment: "production"
</deploy_backend>

<deploy_backend_result>
    status: "success"
    message: "Deployment v1.2.3 to production completed successfully."
    timestamp: "2024-03-15T10:30:00Z"
</deploy_backend_result>

def handle_next_step(thread: Thread):

whileTrue:
    next_step = await determine_next_step(thread_to_prompt(thread))
    
    # 為了清晰起見內聯(lián) - 實際上你可以將其放入一個方法中，使用異常進行控制流，或任何你想要的方式
    if next_step.intent == 'request_clarification':
      thread.events.append({
        type: 'request_clarification',
          data: nextStep,
        })

      await send_message_to_human(next_step)
      await db.save_thread(thread)
      # 異步步驟 - 跳出循環(huán)，我們稍后會收到一個webhook
      break
    elif next_step.intent == 'fetch_open_issues':
      thread.events.append({
        type: 'fetch_open_issues',
        data: next_step,
      })

      issues = await linear_client.issues()

      thread.events.append({
        type: 'fetch_open_issues_result',
        data: issues,
      })
      # 同步步驟 - 將新上下文傳遞給LLM以確定下一步
      continue
    elif next_step.intent == 'create_issue':
      thread.events.append({
        type: 'create_issue',
        data: next_step,
      })

      await request_human_approval(next_step)
      await db.save_thread(thread)
      # 異步步驟 - 跳出循環(huán)，我們稍后會收到一個webhook
      break

thread = {"events": [initial_message]}

whileTrue:
  next_step = await determine_next_step(thread_to_prompt(thread))
  thread["events"].append({
    "type": next_step.intent,
    "data": next_step,
  })
try:
    result = await handle_next_step(thread, next_step) # 我們的switch語句
except Exception as e:
    # 如果出現(xiàn)錯誤，可以將其添加到上下文窗口并重試
    thread["events"].append({
      "type": 'error',
      "data": format_error(e),
    })
    # 循環(huán)，或在此處執(zhí)行其他操作以嘗試恢復

consecutive_errors = 0

whileTrue:

# ... 現(xiàn)有代碼 ...

try:
    result = await handle_next_step(thread, next_step)
    thread["events"].append({
      "type": next_step.intent + '_result',
      data: result,
    })
    # 成功！重置錯誤計數(shù)器
    consecutive_errors = 0
except Exception as e:
    consecutive_errors += 1
    if consecutive_errors < 3:
      # 進行循環(huán)并重試
      thread["events"].append({
        "type": 'error',
        "data": format_error(e),
      })
    else:
      # 中斷循環(huán)，重置上下文窗口的部分內容，升級到人工，或執(zhí)行其他操作
      break
  }
}