一、封装两个基本函数：模型接入、流式输出，以及工具的配置

import openai
from openai import OpenAI
import os
from datetime import datetime
import random
import json
import time
import math
from PIL import Image
#deepseek-chat:deepseek-v3-250324
#deepseek-reasoner:deepseek-r1-250120  (有思考，工具调用，流式输出，结构化输出，不支持查看图片)
#doubao-seed-1.6-thinking(有思考丰富全面，回答丰富有质感,没有工具调用，支持图片)
#qwen-vl-max-latest(没有思考，有工具调用，支持图片，回答的内容比豆包稍逊色些)

#模型接入
def get_response(messages):
    openai_api_key = ""#你的秘钥

    client=OpenAI(api_key=openai_api_key,base_url="")

    response=client.chat.completions.create(
        model="deepseek-r1-250120",  #qwen-vl-max-latest   doubao-seed-1.6-thinking
        messages=messages,
        
        extra_body={
            "enable_thinking": True,
            "thinking_budget": 50
        },
        stream=True,
        tools=tools,
        tool_choice="auto"
        #response_format={"type": "json_object"}
        
    )
    return response
 
 # 初始化一个message数组
messages=[
        {"role": "system", "content": "你是一名阿里云百炼手机商店的店员，你负责给用户推荐手机。手机有两个参数：屏幕尺寸（包括6.1英寸、6.5英寸、6.7英寸）、分辨率（包括2K、4K）。"},
        {"role":"user","content":"介绍一下你知道的手机"}
]

print(get_response(messages).choices[0].message.content)

#流式输出
def stream_output(chunk):
    #修改全局变量时必须用 global
    global reasoning_content
    global answer_content
    global is_answering 
    global is_tool
    global function_name 
    global arguments
    global tool_call_id
    #print(chunk.choices[0].delta)
    
    delta = chunk.choices[0].delta
     # 只收集思考内容
    if hasattr(delta,"reasoning_content") and delta.reasoning_content is not None:
        print(delta.reasoning_content,end="",flush=True)
        reasoning_content +=delta.reasoning_content

    #是否调用工具
    if hasattr(delta, "tool_calls") and delta.tool_calls is not None:
        if not is_tool:
            print("\n" + "=" * 20 + "工具信息" + "=" * 20 + "\n")
            function_name = delta.tool_calls[0].function.name
            tool_call_id = delta.tool_calls[0].id
            is_tool=True
        print(delta.tool_calls[0].function.arguments,end="",flush=True)
        arguments+=delta.tool_calls[0].function.arguments
     # 收到content，开始进行回复
    if hasattr(delta, "content") and delta.content:
        if not is_answering:
            print("\n" + "=" * 20 + "完整回复" + "=" * 20 + "\n")
            is_answering = True
        print(delta.content, end="", flush=True)
        answer_content += delta.content 

#天气函数
def get_current_weather(arguments):
    weather_conditions = [
        "晴天",
        "多云",
        "阴天",
        "雨天",
        "雪天"
    ]
    random_weather = random.choice(weather_conditions)
    location = arguments["location"]
    return f"今天{location}的天气是{random_weather}"

#时间函数
def get_current_time():
    current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    return f"当前时间是{current_time}"

#print(get_current_weather({"location":"北京"}))
#print(get_current_time())
tools =[
    {
        "type": "function",
        "function": {
            "name": "get_current_time",
            "description": "当你想知道现在的时间时非常有用。",
        }
    },
    {
        "type":"function",
        "function":{
            'name':'get_current_weather',
            'description':'当你想查询指定城市的天气时非常有用',
            'parameters':{
                'type':'object',
                'properties':{
                    'location':{
                        'description':"城市或县区，比如北京市、杭州市、余杭区等。",
                        'type':'string'
                    }
                },
                'required':['location']
            }
        }
    }
]

二、工具调用+流式输出+深度思考的完整调用代码

messages1=[
        {"role": "system", "content": "你是一个很有帮助的助手。如果用户提问关于天气的问题，请调用 ‘get_current_weather’ 函数;如果用户提问关于时间的问题，请调用‘get_current_time’函数。请以友好的语气回答问题，根据函数返回的结果推荐适宜的活动。"},
        {"role":"user","content":input("请输入：")}
]

reasoning_content = "" #完整的思考过程
answer_content = "" #完整回复
is_answering = False # 是否进入回复阶段
is_tool=False
function_name =""
arguments=""    #拼接函数参数
tool_call_id=""
print("\n" + "=" * 20 + "思考过程" + "=" * 20 + "\n")

for chunk in get_response(messages1):
    if not chunk.choices:
        print("\nUsage:")
        print(chunk.usage)
        continue
    #print(chunk.choices[0].delta)
    stream_output(chunk)

# 创建一个函数映射表
function_mapper = {
    "get_current_weather": get_current_weather,
    "get_current_time": get_current_time
}
# 获取函数实体
function = function_mapper[function_name]
#如果入参为空，则调用函数
if arguments=="{}":
    function_output = function()
else:
    function_output = function(json.loads(arguments))
print(f"工具函数输出：{function_output}")

messages1.append({"role":"assistant","content":reasoning_content})
#messages中拼接外部函数输出结果
messages1.append(
    {
        "role":"tool",
        "content":function_output,
        "tool_call_id":tool_call_id
    }
)
time.sleep(120)
reasoning_content = "" #完整的思考过程
answer_content = "" #完整回复
is_answering = False # 是否进入回复阶段
is_tool=False
function_name =""
arguments=""    #拼接函数参数
tool_call_id=""
print("\n" + "=" * 20 + "思考过程" + "=" * 20 + "\n")

for chunk in get_response(messages1):
    if not chunk.choices:
        print("\nUsage:")
        print(chunk.usage)
        continue
    #print(chunk.choices[0].delta)
    stream_output(chunk)

三、结构化输出

# 预定义示例响应（用于few-shot提示）
example1_response = json.dumps(
    {
        "info": {"name": "张三", "age": "25岁", "email": "zhangsan@example.com"},
        "hobby": ["唱歌"]
    },
    ensure_ascii=False
)
example2_response = json.dumps(
    {
        "info": {"name": "李四", "age": "30岁", "email": "lisi@example.com"},
        "hobby": ["跳舞", "游泳"]
    },
    ensure_ascii=False
)
messages2=[
        {
            "role": "system",
            "content": f提取name、age、email和hobby（数组类型），输出包含info层和hobby数组的JSON。
            示例：
            Q：我叫张三，今年25岁，邮箱是zhangsan@example.com，爱好是唱歌
            A：{example1_response}
            
            Q：我叫李四，今年30岁，邮箱是lisi@example.com，平时喜欢跳舞和游泳
            A：{example2_response}
            
        },
        {
            "role": "user",
            "content": "大家好，我叫刘五，今年34岁，邮箱是liuwu@example.com，平时喜欢打篮球和旅游", 
        }
    ]
reasoning_content = "" #完整的思考过程
answer_content = "" #完整回复
is_answering = False # 是否进入回复阶段
is_tool=False
function_name =""
arguments=""    #拼接函数参数
tool_call_id=""
print("\n" + "=" * 20 + "思考过程" + "=" * 20 + "\n")

for chunk in get_response(messages2):
    if not chunk.choices:
        print("\nUsage:")
        print(chunk.usage)
        continue
    #print(chunk.choices[0].delta)
    stream_output(chunk)

四、视觉理解：图片解析

messages3=[
        {
            "role": "system",
            "content": "You are a helpful assistant."
            #"content": "You are an AI specialized in recognizing and extracting text from images. Your mission is to analyze the image document and generate the result in QwenVL Document Parser HTML format using specified tags while maintaining user privacy and data integrity." #文档解析
        },
        {
            "role": "user",
            "content": [
                {
                    "type": "image_url",
                    "image_url": {
                        #"url": "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20250212/earbrt/vcg_VCG211286867973_RF.jpg"
                        #"url": "https://img.alicdn.com/imgextra/i2/O1CN01e99Hxt1evMlWM6jUL_!!6000000003933-0-tps-1294-760.jpg"  #做题：qwen-vl-max-latest算的不对   doubao-seed-1.6-thinking算的对
                        "url": "http://duguang-labelling.oss-cn-shanghai.aliyuncs.com/demo_ocr/receipt_zh_demo.jpg"  #信息提取
                        #"url": "https://img.alicdn.com/imgextra/i3/O1CN01I1CXf21UR0Ld20Yzs_!!6000000002513-2-tps-1024-1024.png"  #物体定位
                        #"url": "https://img.alicdn.com/imgextra/i1/O1CN01ILRlNK1gvU5xqbaxb_!!6000000004204-49-tps-1138-640.webp"  #物体定位
                        #"url": "https://img.alicdn.com/imgextra/i3/O1CN01nVbWzy1vx3iInC3z0_!!6000000006238-0-tps-1430-2022.jpg"  #文档解析
                    },
                },
                #{
                #    "type": "video_url",
                #    "video_url": {
                #        "url": "https://cloud.video.taobao.com/vod/C6gCj5AJ3Qrd_UQ9kaMVRY9Ig9G-WToxVYSPRdNXCao.mp4",
                #        "fps": 8.0
                #    },
                #},
                #{"type": "text", "text": "描述一下这副图片的内容"},
                #{"type": "text", "text": "请你分步骤解答这道题，并输出对这道题的思考判断过程"},
                {"type": "text", "text": "提取图中的：['发票代码','发票号码','到站','燃油费','票价','乘车日期','开车时间','车次','座号']，请你以JSON格式输出，不要输出```json```代码段”。"},
                #{"type": "text", "text": "用一个个框定位图像每一个蛋糕的位置并描述其各自的特征，以JSON格式输出所有的bbox的坐标，不要输出```json```代码段"},
                #{"type": "text", "text": "以点的形式定位图中见义勇为的人并详细描述其特征，并以JSON格式输出结果，不要输出```xml```代码段。"},
                #{"type": "text", "text": "请解析一下这篇文档并将英文翻译成中文"},
                #{"type": "text", "text": "请你描述下视频中的人物的一系列动作，按照时间顺序以JSON格式输出开始时间（start_time）、结束事件（end_time）、事件（event），请使用HH:mm:ss表示 时间戳，不要输出```json```代码段。"},
            ],
        },
    ]
reasoning_content = "" #完整的思考过程
answer_content = "" #完整回复
is_answering = False # 是否进入回复阶段
is_tool=False
function_name =""
arguments=""    #拼接函数参数
tool_call_id=""
print("\n" + "=" * 20 + "思考过程" + "=" * 20 + "\n")

for chunk in get_response(messages3):
    if not chunk.choices:
        print("\nUsage:")
        print(chunk.usage)
        continue
    #print(chunk.choices[0].delta)
    stream_output(chunk)

messages3.append({
    "role": "assistant",
    "content": answer_content
})
messages3.append({
    "role": "user",
    "content": "做一首诗描述这个场景"
})

time.sleep(120)
reasoning_content = "" #完整的思考过程
answer_content = "" #完整回复
is_answering = False # 是否进入回复阶段
is_tool=False
function_name =""
arguments=""    #拼接函数参数
tool_call_id=""
print("\n" + "=" * 20 + "思考过程" + "=" * 20 + "\n")

for chunk in get_response(messages3):
    if not chunk.choices:
        print("\nUsage:")
        print(chunk.usage)
        continue
    #print(chunk.choices[0].delta)
    stream_output(chunk)

五、多轮对话

messages=[
        {"role": "system", "content": "你是一名阿里云百炼手机商店的店员，你负责给用户推荐手机。手机有两个参数：屏幕尺寸（包括6.1英寸、6.5英寸、6.7英寸）、分辨率（包括2K、4K）。你一次只能向用户提问一个参数。如果用户提供的信息不全，你需要反问他，让他提供没有提供的参数。如果参数收集完成，你要说：我已了解您的购买意向，请稍等。"}
]

full_content = "欢迎光临阿里云百炼手机商店，您需要购买什么尺寸的手机呢？"

print(f"模型输出：{full_content}\n")
while "我已了解您的购买意向" not in full_content:
    user_input=input("请输入：")
    #将用户问题信息添加到messages列表中
    messages.append({"role":"user","content":user_input})
    for chunk in get_response(messages):
    # 如果stream_options.include_usage为True，则最后一个chunk的choices字段为空列表，需要跳过（可以通过chunk.usage获取 Token 使用量）
        if chunk.choices:
            full_content += chunk.choices[0].delta.content
            print(chunk.choices[0].delta.content, end="", flush=True)
    #assistant_output = get_response(messages).choices[0].message.content
    #将大模型的回复信息添加到messages列表中
    messages.append({"role":"assistant","content":full_content})
   # print(f"模型输出：{assistant_output}")
    print("\n")
# 语言模型 # 人工智能
Logo
亚马逊云科技技术品牌专区

更多推荐

open-code-review Codex 插件教程：对话式调用评审技能指南

本文是一份面向新手的 **open-code-review Codex 插件教程**，教你在 Codex 中通过对话式方式调用 open-code-review 的 AI 代码评审技能。open-code-review 是阿里巴巴开源的 AI 代码评审工具，采用"确定性流水线 + LLM Agent"混合架构，内置多语言评审规则（NPE、线程安全、XSS、SQL 注入等），可在 Codex 里用一

avatar 亚马逊云科技技术品牌专区

OBS Studio 完整指南：免费开源录屏与直播软件从零上手

想做一场像模像样的直播，或者把屏幕内容录成教程视频，但一想到商用软件的订阅费就有些犹豫？其实你不需要花一分钱。OBS Studio 是一款免费开源的直播与录屏软件，支持 Windows、macOS 和 Linux，采集、混音、编码、推流、录制，一条流水线全包了。## 先搞清楚 OBS Studio 是什么OBS Studio 是一个"场景化"的采集合成工具：你把摄像头、窗口、浏览器、图片、

avatar 亚马逊云科技技术品牌专区

如何用MediaCrawler爬取小红书抖音数据？7大平台爬虫从安装到实战完整指南

MediaCrawler 是一个支持小红书、抖音、快手、B站、微博、百度贴吧、知乎的开源自媒体数据采集工具。它能按关键词、按帖子、按创作者主页三种方式抓取笔记/视频内容、评论甚至二级评论，并把结果存成 JSON、CSV、Excel 或数据库。适合做市场调研、竞品监控、内容选题的运营人员，以及想要现成爬虫框架而不想从零写 JS 逆向的开发者。## 从一次"手动抄数据"说起想象这个场景：你负责

avatar 亚马逊云科技技术品牌专区